← Back
Editing: clickhouse.lua
--[[ Copyright (c) 2022, Vsevolod Stakhov <vsevolod@rspamd.com> Licensed 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. ]]-- local rspamd_logger = require 'rspamd_logger' local upstream_list = require "rspamd_upstream_list" local lua_util = require "lua_util" local lua_clickhouse = require "lua_clickhouse" local lua_settings = require "lua_settings" local fun = require "fun" local N = "clickhouse" if confighelp then return end local data_rows = {} local custom_rows = {} local nrows = 0 local used_memory = 0 local last_collection = 0 local final_call = false -- If the final collection has been started local schema_version = 11 -- Current schema version local extra_tables = {} local extra_table_rows = {} local settings = { limits = { -- Collection limits max_rows = 1000, -- How many rows are allowed (0 for disable this) max_memory = 50 * 1024 * 1024, -- How many memory should be occupied before sending collection max_interval = 60, -- Maximum collection interval }, collect_garbage = false, -- Perform GC collection after sending the data check_timeout = 10.0, -- Periodic timeout timeout = 5.0, bayes_spam_symbols = { 'BAYES_SPAM' }, bayes_ham_symbols = { 'BAYES_HAM' }, ann_symbols_spam = { 'NEURAL_SPAM' }, ann_symbols_ham = { 'NEURAL_HAM' }, fuzzy_symbols = { 'FUZZY_DENIED' }, whitelist_symbols = { 'WHITELIST_DKIM', 'WHITELIST_SPF_DKIM', 'WHITELIST_DMARC' }, dkim_allow_symbols = { 'R_DKIM_ALLOW' }, dkim_reject_symbols = { 'R_DKIM_REJECT' }, dkim_dnsfail_symbols = { 'R_DKIM_TEMPFAIL', 'R_DKIM_PERMFAIL' }, dkim_na_symbols = { 'R_DKIM_NA' }, dmarc_allow_symbols = { 'DMARC_POLICY_ALLOW' }, dmarc_reject_symbols = { 'DMARC_POLICY_REJECT' }, dmarc_quarantine_symbols = { 'DMARC_POLICY_QUARANTINE' }, dmarc_softfail_symbols = { 'DMARC_POLICY_SOFTFAIL' }, dmarc_na_symbols = { 'DMARC_NA' }, spf_allow_symbols = { 'R_SPF_ALLOW' }, spf_reject_symbols = { 'R_SPF_FAIL' }, spf_dnsfail_symbols = { 'R_SPF_DNSFAIL', 'R_SPF_PERMFAIL' }, spf_neutral_symbols = { 'R_DKIM_TEMPFAIL', 'R_DKIM_PERMFAIL' }, spf_na_symbols = { 'R_SPF_NA' }, stop_symbols = {}, ipmask = 19, ipmask6 = 48, full_urls = false, from_tables = nil, enable_symbols = false, database = 'default', use_https = false, use_gzip = true, allow_local = false, insert_subject = false, subject_privacy = false, -- subject privacy is off subject_privacy_alg = 'blake2', -- default hash-algorithm to obfuscate subject subject_privacy_prefix = 'obf', -- prefix to show it's obfuscated subject_privacy_length = 16, -- cut the length of the hash schema_additions = {}, -- additional SQL statements to be executed when schema is uploaded user = nil, password = nil, no_ssl_verify = false, custom_rules = {}, enable_digest = false, enable_uuid = true, exceptions = nil, retention = { enable = false, method = 'detach', period_months = 3, run_every = '7d', }, extra_columns = {}, -- Inject a curated set of extra_columns from a named preset. Currently -- supported: "outbound" (per-sender outbound traffic profiling columns: -- languages, hour-of-day, envelope-from domain, recipient domains, etc.). -- User-supplied entries with the same name take precedence. Schema -- migration reuses the existing extra_columns path. Unset by default. preset = nil, } -- Named dispatch table for curated `extra_columns` presets. Each preset is an -- array of entries using the same shape as user-supplied `extra_columns` -- entries (see processing below). Selectors in these tables must be -- expressible with the stock selectors that ship in rspamd master. Additional -- presets (e.g. "inbound", "compliance") can be added as new keys here. local extra_column_presets = { outbound = { { name = 'Languages', type = 'Array(LowCardinality(String))', selector = 'languages', default_value = {}, comment = 'outbound preset: detected languages of text parts', }, { name = 'ReceivedCount', type = 'UInt16', selector = 'received_count', default_value = '0', comment = 'outbound preset: number of Received headers (hop count)', }, { name = 'FuzzyDigest', type = 'String', selector = 'fuzzy_digest', default_value = '', comment = 'outbound preset: strong fuzzy digest of largest text part (for fan-out aggregations)', }, }, } --- @language SQL local clickhouse_schema = { [[ CREATE TABLE IF NOT EXISTS rspamd ( Date Date COMMENT 'Date (used for partitioning)', TS DateTime COMMENT 'Date and time of the request start (UTC)', From String COMMENT 'Domain part of the return address (RFC5321.MailFrom)', MimeFrom String COMMENT 'Domain part of the address in From: header (RFC5322.From)', IP String COMMENT 'SMTP client IP as provided by MTA or from Received: header', Helo String COMMENT 'Full hostname as sent by the SMTP client (RFC5321.HELO/.EHLO)', Score Float32 COMMENT 'Message score', NRcpt UInt8 COMMENT 'Number of envelope recipients (RFC5321.RcptTo)', Size UInt32 COMMENT 'Message size in bytes', IsWhitelist Enum8('blacklist' = 0, 'whitelist' = 1, 'unknown' = 2) DEFAULT 'unknown' COMMENT 'Based on symbols configured in `whitelist_symbols` module option', IsBayes Enum8('ham' = 0, 'spam' = 1, 'unknown' = 2) DEFAULT 'unknown' COMMENT 'Based on symbols configured in `bayes_spam_symbols` and `bayes_ham_symbols` module options', IsFuzzy Enum8('whitelist' = 0, 'deny' = 1, 'unknown' = 2) DEFAULT 'unknown' COMMENT 'Based on symbols configured in `fuzzy_symbols` module option', IsFann Enum8('ham' = 0, 'spam' = 1, 'unknown' = 2) DEFAULT 'unknown' COMMENT 'Based on symbols configured in `ann_symbols_spam` and `ann_symbols_ham` module options', IsDkim Enum8('reject' = 0, 'allow' = 1, 'unknown' = 2, 'dnsfail' = 3, 'na' = 4) DEFAULT 'unknown' COMMENT 'Based on symbols configured in dkim_* module options', IsDmarc Enum8('reject' = 0, 'allow' = 1, 'unknown' = 2, 'softfail' = 3, 'na' = 4, 'quarantine' = 5) DEFAULT 'unknown' COMMENT 'Based on symbols configured in dmarc_* module options', IsSpf Enum8('reject' = 0, 'allow' = 1, 'neutral' = 2, 'dnsfail' = 3, 'na' = 4, 'unknown' = 5) DEFAULT 'unknown' COMMENT 'Based on symbols configured in spf_* module options', NUrls Int32 COMMENT 'Number of URLs and email extracted from the message', Action Enum8('reject' = 0, 'rewrite subject' = 1, 'add header' = 2, 'greylist' = 3, 'no action' = 4, 'soft reject' = 5, 'custom' = 6) DEFAULT 'no action' COMMENT 'Action returned for the message; if action is not predefined actual action will be in `CustomAction` field', CustomAction LowCardinality(String) COMMENT 'Action string for custom action', FromUser String COMMENT 'Local part of the return address (RFC5321.MailFrom)', MimeUser String COMMENT 'Local part of the address in From: header (RFC5322.From)', RcptUser String COMMENT '[Deprecated] Local part of the first envelope recipient (RFC5321.RcptTo)', RcptDomain String COMMENT '[Deprecated] Domain part of the first envelope recipient (RFC5321.RcptTo)', SMTPRecipients Array(String) COMMENT 'List of envelope recipients (RFC5321.RcptTo)', MimeRecipients Array(String) COMMENT 'List of recipients from headers (RFC5322.To/.CC/.BCC)', MessageId String COMMENT 'Message-ID header', ListId String COMMENT 'List-Id header', Subject String COMMENT 'Subject header (or hash if `subject_privacy` module option enabled)', `Attachments.FileName` Array(String) COMMENT 'Attachment name', `Attachments.ContentType` Array(String) COMMENT 'Attachment Content-Type', `Attachments.Length` Array(UInt32) COMMENT 'Attachment size in bytes', `Attachments.Digest` Array(FixedString(16)) COMMENT 'First 16 characters of hash returned by mime_part:get_digest()', `Urls.Tld` Array(String) COMMENT 'Effective second level domain part of the URL host', `Urls.Url` Array(String) COMMENT 'Full URL if `full_urls` module option enabled, host part of URL otherwise', `Urls.Flags` Array(UInt32) COMMENT 'Corresponding url flags, see `enum rspamd_url_flags` in libserver/url.h for details', Emails Array(String) COMMENT 'List of emails extracted from the message', `Fuzzy.Hash` Array(String) COMMENT 'Stored fuzzy digest that matched', `Fuzzy.Queried` Array(String) COMMENT 'Fuzzy digest of the message part that was queried', `Fuzzy.Rule` Array(LowCardinality(String)) COMMENT 'Fuzzy rule name', `Fuzzy.Prob` Array(Float32) COMMENT 'Fuzzy match probability (shingle overlap ratio)', `Fuzzy.Flag` Array(UInt32) COMMENT 'Fuzzy storage flag', ASN UInt32 COMMENT 'BGP AS number for SMTP client IP (returned by asn.rspamd.com or asn6.rspamd.com)', Country FixedString(2) COMMENT 'Country for SMTP client IP (returned by asn.rspamd.com or asn6.rspamd.com)', IPNet String, `Symbols.Names` Array(LowCardinality(String)) COMMENT 'Symbol name', `Symbols.Scores` Array(Float32) COMMENT 'Symbol score', `Symbols.Options` Array(String) COMMENT 'Symbol options (comma separated list)', `Groups.Names` Array(LowCardinality(String)) COMMENT 'Group name', `Groups.Scores` Array(Float32) COMMENT 'Group score', ScanTimeReal UInt32 COMMENT 'Request time in milliseconds', ScanTimeVirtual UInt32 COMMENT 'Deprecated do not use', AuthUser String COMMENT 'Username for authenticated SMTP client', SettingsId LowCardinality(String) COMMENT 'ID for the settings profile', Digest FixedString(32) COMMENT '[Deprecated]', TaskUUID UUID COMMENT 'Native UUID v7 (RFC 9562) for task identification', SMTPFrom ALIAS if(From = '', '', concat(FromUser, '@', From)) COMMENT 'Return address (RFC5321.MailFrom)', SMTPRcpt ALIAS SMTPRecipients[1] COMMENT 'The first envelope recipient (RFC5321.RcptTo)', MIMEFrom ALIAS if(MimeFrom = '', '', concat(MimeUser, '@', MimeFrom)) COMMENT 'Address in From: header (RFC5322.From)', MIMERcpt ALIAS MimeRecipients[1] COMMENT 'The first recipient from headers (RFC5322.To/.CC/.BCC)' ) ENGINE = MergeTree() PARTITION BY toMonday(Date) ORDER BY TS ]], [[CREATE TABLE IF NOT EXISTS rspamd_version ( Version UInt32) ENGINE = TinyLog]], { [[INSERT INTO rspamd_version (Version) Values (${SCHEMA_VERSION})]], true }, } -- This describes SQL queries to migrate between versions local migrations = { [1] = { -- Move to a wide fat table [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS `Attachments.FileName` Array(String) AFTER ListId, ADD COLUMN IF NOT EXISTS `Attachments.ContentType` Array(String) AFTER `Attachments.FileName`, ADD COLUMN IF NOT EXISTS `Attachments.Length` Array(UInt32) AFTER `Attachments.ContentType`, ADD COLUMN IF NOT EXISTS `Attachments.Digest` Array(FixedString(16)) AFTER `Attachments.Length`, ADD COLUMN IF NOT EXISTS `Urls.Tld` Array(String) AFTER `Attachments.Digest`, ADD COLUMN IF NOT EXISTS `Urls.Url` Array(String) AFTER `Urls.Tld`, ADD COLUMN IF NOT EXISTS Emails Array(String) AFTER `Urls.Url`, ADD COLUMN IF NOT EXISTS ASN UInt32 AFTER Emails, ADD COLUMN IF NOT EXISTS Country FixedString(2) AFTER ASN, ADD COLUMN IF NOT EXISTS IPNet String AFTER Country, ADD COLUMN IF NOT EXISTS `Symbols.Names` Array(String) AFTER IPNet, ADD COLUMN IF NOT EXISTS `Symbols.Scores` Array(Float64) AFTER `Symbols.Names`, ADD COLUMN IF NOT EXISTS `Symbols.Options` Array(String) AFTER `Symbols.Scores`]], -- Add explicit version [[CREATE TABLE rspamd_version ( Version UInt32) ENGINE = TinyLog]], [[INSERT INTO rspamd_version (Version) Values (2)]], }, [2] = { -- Add `Subject` column [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS Subject String AFTER ListId]], -- New version [[INSERT INTO rspamd_version (Version) Values (3)]], }, [3] = { [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS IsSpf Enum8('reject' = 0, 'allow' = 1, 'neutral' = 2, 'dnsfail' = 3, 'na' = 4, 'unknown' = 5) DEFAULT 'unknown' AFTER IsDmarc, MODIFY COLUMN IsDkim Enum8('reject' = 0, 'allow' = 1, 'unknown' = 2, 'dnsfail' = 3, 'na' = 4) DEFAULT 'unknown', MODIFY COLUMN IsDmarc Enum8('reject' = 0, 'allow' = 1, 'unknown' = 2, 'softfail' = 3, 'na' = 4, 'quarantine' = 5) DEFAULT 'unknown', ADD COLUMN IF NOT EXISTS MimeRecipients Array(String) AFTER RcptDomain, ADD COLUMN IF NOT EXISTS MessageId String AFTER MimeRecipients, ADD COLUMN IF NOT EXISTS ScanTimeReal UInt32 AFTER `Symbols.Options`, ADD COLUMN IF NOT EXISTS ScanTimeVirtual UInt32 AFTER ScanTimeReal]], -- Add aliases [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS SMTPFrom ALIAS if(From = '', '', concat(FromUser, '@', From)), ADD COLUMN IF NOT EXISTS SMTPRcpt ALIAS if(RcptDomain = '', '', concat(RcptUser, '@', RcptDomain)), ADD COLUMN IF NOT EXISTS MIMEFrom ALIAS if(MimeFrom = '', '', concat(MimeUser, '@', MimeFrom)), ADD COLUMN IF NOT EXISTS MIMERcpt ALIAS MimeRecipients[1] ]], -- New version [[INSERT INTO rspamd_version (Version) Values (4)]], }, [4] = { [[ALTER TABLE rspamd MODIFY COLUMN Action Enum8('reject' = 0, 'rewrite subject' = 1, 'add header' = 2, 'greylist' = 3, 'no action' = 4, 'soft reject' = 5, 'custom' = 6) DEFAULT 'no action', ADD COLUMN IF NOT EXISTS CustomAction String AFTER Action ]], -- New version [[INSERT INTO rspamd_version (Version) Values (5)]], }, [5] = { [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS AuthUser String AFTER ScanTimeVirtual, ADD COLUMN IF NOT EXISTS SettingsId LowCardinality(String) AFTER AuthUser ]], -- New version [[INSERT INTO rspamd_version (Version) Values (6)]], }, [6] = { -- Add new columns [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS Helo String AFTER IP, ADD COLUMN IF NOT EXISTS SMTPRecipients Array(String) AFTER RcptDomain ]], -- Modify SMTPRcpt alias [[ ALTER TABLE rspamd MODIFY COLUMN SMTPRcpt ALIAS SMTPRecipients[1] ]], -- New version [[INSERT INTO rspamd_version (Version) Values (7)]], }, [7] = { -- Add new columns [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS `Groups.Names` Array(LowCardinality(String)) AFTER `Symbols.Options`, ADD COLUMN IF NOT EXISTS `Groups.Scores` Array(Float32) AFTER `Groups.Names` ]], -- New version [[INSERT INTO rspamd_version (Version) Values (8)]], }, [8] = { -- Add new columns [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS `Urls.Flags` Array(UInt32) AFTER `Urls.Url` ]], -- New version [[INSERT INTO rspamd_version (Version) Values (9)]], }, [9] = { -- Add UUID column [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS TaskUUID UUID AFTER Digest ]], -- New version [[INSERT INTO rspamd_version (Version) Values (10)]], }, [10] = { -- Add fuzzy matches information [[ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS `Fuzzy.Hash` Array(String) AFTER Emails, ADD COLUMN IF NOT EXISTS `Fuzzy.Queried` Array(String) AFTER `Fuzzy.Hash`, ADD COLUMN IF NOT EXISTS `Fuzzy.Rule` Array(LowCardinality(String)) AFTER `Fuzzy.Queried`, ADD COLUMN IF NOT EXISTS `Fuzzy.Prob` Array(Float32) AFTER `Fuzzy.Rule`, ADD COLUMN IF NOT EXISTS `Fuzzy.Flag` Array(UInt32) AFTER `Fuzzy.Prob` ]], -- New version [[INSERT INTO rspamd_version (Version) Values (11)]], }, } local predefined_actions = { ['reject'] = true, ['rewrite subject'] = true, ['add header'] = true, ['greylist'] = true, ['no action'] = true, ['soft reject'] = true } local function clickhouse_main_row(res) local fields = { 'Date', 'TS', 'From', 'MimeFrom', 'IP', 'Helo', 'Score', 'NRcpt', 'Size', 'IsWhitelist', 'IsBayes', 'IsFuzzy', 'IsFann', 'IsDkim', 'IsDmarc', 'NUrls', 'Action', 'FromUser', 'MimeUser', 'RcptUser', 'RcptDomain', 'SMTPRecipients', 'ListId', 'Subject', 'Digest', 'TaskUUID', -- 1.9.2 + 'IsSpf', 'MimeRecipients', 'MessageId', 'ScanTimeReal', -- 1.9.3 + 'CustomAction', -- 2.0 + 'AuthUser', 'SettingsId', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_attachments_row(res) local fields = { 'Attachments.FileName', 'Attachments.ContentType', 'Attachments.Length', 'Attachments.Digest', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_urls_row(res) local fields = { 'Urls.Tld', 'Urls.Url', 'Urls.Flags', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_emails_row(res) local fields = { 'Emails', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_fuzzy_row(res) local fields = { 'Fuzzy.Hash', 'Fuzzy.Queried', 'Fuzzy.Rule', 'Fuzzy.Prob', 'Fuzzy.Flag', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_symbols_row(res) local fields = { 'Symbols.Names', 'Symbols.Scores', 'Symbols.Options', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_groups_row(res) local fields = { 'Groups.Names', 'Groups.Scores', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_asn_row(res) local fields = { 'ASN', 'Country', 'IPNet', } for _, v in ipairs(fields) do table.insert(res, v) end end local function clickhouse_extra_columns(res) for _, v in ipairs(settings.extra_columns) do table.insert(res, v.name) end end local function today(ts) return os.date('!%Y-%m-%d', ts) end local function clickhouse_check_symbol(task, settings_field_name, fields_table, field_name, value, value_negative) for _, s in ipairs(settings[settings_field_name] or {}) do if task:has_symbol(s) then if value_negative then local sym = task:get_symbol(s)[1] if sym['score'] > 0 then fields_table[field_name] = value else fields_table[field_name] = value_negative end else fields_table[field_name] = value end return true end end return false end local function upload_extra_table_schema(upstream, ev_base, cfg, tbl_config) local ch_params = { ev_base = ev_base, config = cfg, } local sql = lua_util.template(tbl_config.schema, settings) local err, _ = lua_clickhouse.generic_sync(upstream, settings, ch_params, sql) if err then return err end rspamd_logger.infox(rspamd_config, 'uploaded extra table schema for %s (table: %s)', tbl_config.name, tbl_config.table_name) return nil end local function clickhouse_send_data(task, ev_base, why, gen_rows, cust_rows, extra_rows) local log_object = task or rspamd_config local upstream = settings.upstream:get_upstream_round_robin() if not upstream then rspamd_logger.errx(log_object, "no clickhouse upstream available (DNS pending or all dead); skipping send (%s)", why) return end local ip_addr = upstream:get_addr():to_string(true) rspamd_logger.infox(log_object, "trying to send %s rows to clickhouse server %s; started as %s", #gen_rows + #cust_rows, ip_addr, why) local function gen_success_cb(what, how_many) return function(_, _) rspamd_logger.messagex(log_object, "sent %s rows of %s to clickhouse server %s; started as %s", how_many, what, ip_addr, why) upstream:ok() end end local function gen_fail_cb(what, how_many) return function(_, err) rspamd_logger.errx(log_object, "cannot send %s rows of %s data to clickhouse server %s: %s; started as %s", how_many, what, ip_addr, err, why) upstream:fail() end end local function send_data(what, tbl, query) local ch_params = {} if task then ch_params.task = task else ch_params.config = rspamd_config ch_params.ev_base = ev_base end local ret = lua_clickhouse.insert(upstream, settings, ch_params, query, tbl, gen_success_cb(what, #tbl), gen_fail_cb(what, #tbl)) if not ret then rspamd_logger.errx(log_object, "cannot send %s rows of %s data to clickhouse server %s: %s", #tbl, what, ip_addr, 'cannot make HTTP request') end end local fields = {} clickhouse_main_row(fields) clickhouse_attachments_row(fields) clickhouse_urls_row(fields) clickhouse_emails_row(fields) clickhouse_fuzzy_row(fields) clickhouse_asn_row(fields) if settings.enable_symbols then clickhouse_symbols_row(fields) clickhouse_groups_row(fields) end if #settings.extra_columns > 0 then clickhouse_extra_columns(fields) end send_data('generic data', gen_rows, string.format('INSERT INTO rspamd (%s)', table.concat(fields, ','))) for k, crows in pairs(cust_rows) do if #crows > 1 then send_data('custom data (' .. k .. ')', crows, settings.custom_rules[k].first_row()) end end if extra_rows then for name, tbl_config in pairs(extra_tables) do local trows = extra_rows[name] if trows and #trows > 0 then if not tbl_config.schema_uploaded then local err = upload_extra_table_schema(upstream, ev_base, rspamd_config, tbl_config) if err then rspamd_logger.errx(log_object, 'failed to upload schema for extra table %s: %s', name, err) else tbl_config.schema_uploaded = true end end if tbl_config.schema_uploaded then send_data('extra table (' .. name .. ')', trows, tbl_config.insert_query()) end end end end end --[[ Register an extra Clickhouse table from another plugin. @param opts table with fields: - name: unique registration identifier - table_name: Clickhouse table name - schema: CREATE TABLE statement - insert_query: function returning INSERT statement - get_row: function(task) returning row data table, array of rows, or nil - retention: optional {enable, period_months, method} @return success, error_message ]] local function register_extra_table(opts) if not settings.upstream then return false, 'clickhouse plugin is not configured' end if not opts.name or type(opts.name) ~= 'string' or #opts.name == 0 then return false, 'name is required and must be a non-empty string' end if not opts.table_name or type(opts.table_name) ~= 'string' or #opts.table_name == 0 then return false, 'table_name is required and must be a non-empty string' end if not opts.schema or type(opts.schema) ~= 'string' or #opts.schema == 0 then return false, 'schema is required and must be a non-empty string' end if type(opts.insert_query) ~= 'function' then return false, 'insert_query must be a function' end if type(opts.get_row) ~= 'function' then return false, 'get_row must be a function' end if extra_tables[opts.name] then return false, string.format('extra table "%s" is already registered', opts.name) end local retention = { enable = false } if opts.retention then if type(opts.retention) ~= 'table' then return false, 'retention must be a table' end retention.enable = opts.retention.enable or false retention.period_months = opts.retention.period_months or settings.retention.period_months retention.method = opts.retention.method or settings.retention.method if retention.method ~= 'drop' and retention.method ~= 'detach' then return false, 'retention.method must be "drop" or "detach"' end if type(retention.period_months) ~= 'number' or retention.period_months < 1 or retention.period_months > 1000 then return false, 'retention.period_months must be between 1 and 1000' end end extra_tables[opts.name] = { name = opts.name, table_name = opts.table_name, schema = opts.schema, insert_query = opts.insert_query, get_row = opts.get_row, retention = retention, schema_uploaded = false, } rspamd_logger.infox(rspamd_config, 'registered clickhouse extra table: %s (table: %s)', opts.name, opts.table_name) return true, nil end local function unregister_extra_table(name) if extra_tables[name] then extra_tables[name] = nil extra_table_rows[name] = nil rspamd_logger.infox(rspamd_config, 'unregistered clickhouse extra table: %s', name) return true end return false end local function get_extra_tables() local result = {} for name, config in pairs(extra_tables) do result[name] = { name = config.name, table_name = config.table_name, schema_uploaded = config.schema_uploaded, retention = config.retention, } end return result end local function clickhouse_collect(task) if task:has_flag('skip') then return end if not settings.allow_local and lua_util.is_rspamc_or_controller(task) then return end for _, sym in ipairs(settings.stop_symbols) do if task:has_symbol(sym) then rspamd_logger.infox(task, 'skip Clickhouse storage for message: symbol %s has fired', sym) return end end if settings.exceptions then local excepted, trace = settings.exceptions:process(task) if excepted then rspamd_logger.infox(task, 'skipped Clickhouse storage for message: excepted (%s)', trace) -- Excepted return end end local from_domain = '' local from_user = '' if task:has_from('smtp') then local from = task:get_from({ 'smtp', 'orig' })[1] if from then from_domain = from['domain']:lower() from_user = from['user'] end end local mime_domain = '' local mime_user = '' if task:has_from('mime') then local from = task:get_from({ 'mime', 'orig' })[1] if from then mime_domain = from['domain']:lower() mime_user = from['user'] end end local mime_recipients = {} if task:has_recipients('mime') then local recipients = task:get_recipients({ 'mime', 'orig' }) for _, rcpt in ipairs(recipients) do table.insert(mime_recipients, rcpt['user'] .. '@' .. rcpt['domain']:lower()) end end local ip_str = 'undefined' local ip = task:get_from_ip() if ip and ip:is_valid() then local ipnet if ip:get_version() == 4 then ipnet = ip:apply_mask(settings['ipmask']) else ipnet = ip:apply_mask(settings['ipmask6']) end ip_str = ipnet:to_string() end local helo = task:get_helo() or '' local rcpt_user = '' local rcpt_domain = '' local smtp_recipients = {} if task:has_recipients('smtp') then local recipients = task:get_recipients('smtp') -- for compatibility with an old table structure rcpt_user = recipients[1]['user'] rcpt_domain = recipients[1]['domain']:lower() for _, rcpt in ipairs(recipients) do table.insert(smtp_recipients, rcpt['user'] .. '@' .. rcpt['domain']:lower()) end end local list_id = task:get_header('List-Id') or '' local message_id = lua_util.maybe_obfuscate_string(task:get_message_id() or '', settings, 'mid') local score = task:get_metric_score()[1]; local fields = { bayes = 'unknown', fuzzy = 'unknown', ann = 'unknown', whitelist = 'unknown', dkim = 'unknown', dmarc = 'unknown', spf = 'unknown', } local ret ret = clickhouse_check_symbol(task, 'bayes_spam_symbols', fields, 'bayes', 'spam') if not ret then clickhouse_check_symbol(task, 'bayes_ham_symbols', fields, 'bayes', 'ham') end clickhouse_check_symbol(task, 'ann_symbols_spam', fields, 'ann', 'spam') if not ret then clickhouse_check_symbol(task, 'ann_symbols_ham', fields, 'ann', 'ham') end clickhouse_check_symbol(task, 'whitelist_symbols', fields, 'whitelist', 'blacklist', 'whitelist') clickhouse_check_symbol(task, 'fuzzy_symbols', fields, 'fuzzy', 'deny') ret = clickhouse_check_symbol(task, 'dkim_allow_symbols', fields, 'dkim', 'allow') if not ret then ret = clickhouse_check_symbol(task, 'dkim_reject_symbols', fields, 'dkim', 'reject') end if not ret then ret = clickhouse_check_symbol(task, 'dkim_dnsfail_symbols', fields, 'dkim', 'dnsfail') end if not ret then clickhouse_check_symbol(task, 'dkim_na_symbols', fields, 'dkim', 'na') end ret = clickhouse_check_symbol(task, 'dmarc_allow_symbols', fields, 'dmarc', 'allow') if not ret then ret = clickhouse_check_symbol(task, 'dmarc_reject_symbols', fields, 'dmarc', 'reject') end if not ret then ret = clickhouse_check_symbol(task, 'dmarc_quarantine_symbols', fields, 'dmarc', 'quarantine') end if not ret then ret = clickhouse_check_symbol(task, 'dmarc_softfail_symbols', fields, 'dmarc', 'softfail') end if not ret then clickhouse_check_symbol(task, 'dmarc_na_symbols', fields, 'dmarc', 'na') end ret = clickhouse_check_symbol(task, 'spf_allow_symbols', fields, 'spf', 'allow') if not ret then ret = clickhouse_check_symbol(task, 'spf_reject_symbols', fields, 'spf', 'reject') end if not ret then ret = clickhouse_check_symbol(task, 'spf_neutral_symbols', fields, 'spf', 'neutral') end if not ret then ret = clickhouse_check_symbol(task, 'spf_dnsfail_symbols', fields, 'spf', 'dnsfail') end if not ret then clickhouse_check_symbol(task, 'spf_na_symbols', fields, 'spf', 'na') end local nrcpts = 0 if task:has_recipients('smtp') then nrcpts = #task:get_recipients('smtp') end local nurls = 0 local task_urls = task:get_urls({ content = true, images = true, emails = false, sort = true, }) or {} nurls = #task_urls local timestamp = math.floor(task:get_date({ format = 'connect', gmt = true, -- The only sane way to sync stuff with different timezones })) local action = task:get_metric_action() local custom_action = '' if not predefined_actions[action] then custom_action = action action = 'custom' end local digest = '' if settings.enable_digest then digest = task:get_digest() end local task_uuid = '' if settings.enable_uuid then task_uuid = task:get_uuid() or '' end local subject = '' if settings.insert_subject then subject = lua_util.maybe_obfuscate_string(task:get_subject() or '', settings, 'subject') end local scan_real = task:get_scan_time() scan_real = math.floor(scan_real * 1000) if scan_real < 0 then rspamd_logger.messagex(task, 'clock skew detected for message: %s ms real scan time (reset to 0)', scan_real) scan_real = 0 end local auth_user = task:get_user() or '' local settings_id = task:get_settings_id() if settings_id then -- Convert to string settings_id = lua_settings.settings_by_id(settings_id) if settings_id then settings_id = settings_id.name end end if not settings_id then settings_id = '' end local row = { today(timestamp), timestamp, from_domain, mime_domain, ip_str, helo, score, nrcpts, task:get_size(), fields.whitelist, fields.bayes, fields.fuzzy, fields.ann, fields.dkim, fields.dmarc, nurls, action, from_user, mime_user, rcpt_user, rcpt_domain, smtp_recipients, list_id, subject, digest, task_uuid, fields.spf, mime_recipients, message_id, scan_real, custom_action, auth_user, settings_id } -- Attachments step local attachments_fnames = {} local attachments_ctypes = {} local attachments_lengths = {} local attachments_digests = {} for _, part in ipairs(task:get_parts()) do if part:is_attachment() then table.insert(attachments_fnames, part:get_filename() or '') local mime_type, mime_subtype = part:get_type() table.insert(attachments_ctypes, string.format("%s/%s", mime_type, mime_subtype)) table.insert(attachments_lengths, part:get_length()) table.insert(attachments_digests, string.sub(part:get_digest(), 1, 16)) end end if #attachments_fnames > 0 then table.insert(row, attachments_fnames) table.insert(row, attachments_ctypes) table.insert(row, attachments_lengths) table.insert(row, attachments_digests) else table.insert(row, {}) table.insert(row, {}) table.insert(row, {}) table.insert(row, {}) end -- Urls step local urls_urls = {} local urls_tlds = {} local urls_flags = {} if settings.full_urls then for i, u in ipairs(task_urls) do urls_urls[i] = u:get_text() urls_tlds[i] = u:get_tld() or u:get_host() urls_flags[i] = u:get_flags_num() end else -- We need to store unique local mt = { ord_tbl = {}, -- ordered list of urls idx_tbl = {}, -- indexed by host + flags, reference to an index in ord_tbl __newindex = function(t, k, v) local idx = getmetatable(t).idx_tbl local ord = getmetatable(t).ord_tbl local key = k:get_host() .. tostring(k:get_flags_num()) if idx[key] then ord[idx[key]] = v -- replace else ord[#ord + 1] = v idx[key] = #ord end end, __index = function(t, k) local ord = getmetatable(t).ord_tbl if type(k) == 'number' then return ord[k] else local idx = getmetatable(t).idx_tbl local key = k:get_host() .. tostring(k:get_flags_num()) if idx[key] then return ord[idx[key]] end end end, } -- Extra index needed for making this unique local urls_idx = {} setmetatable(urls_idx, mt) for _, u in ipairs(task_urls) do if not urls_idx[u] then urls_idx[u] = u urls_urls[#urls_urls + 1] = u:get_host() urls_tlds[#urls_tlds + 1] = u:get_tld() or u:get_host() urls_flags[#urls_flags + 1] = u:get_flags_num() end end end -- Get tlds table.insert(row, urls_tlds) -- Get hosts/full urls table.insert(row, urls_urls) -- Numeric flags table.insert(row, urls_flags) -- Emails step if task:has_urls(true) then local emails = task:get_emails() or {} local emails_formatted = {} for i, u in ipairs(emails) do emails_formatted[i] = string.format('%s@%s', u:get_user(), u:get_host()) end table.insert(row, emails_formatted) else table.insert(row, {}) end -- Fuzzy matches local fuzzy_hashes = {} local fuzzy_queried = {} local fuzzy_rules = {} local fuzzy_probs = {} local fuzzy_flags = {} for _, m in ipairs(task:get_fuzzy_results() or {}) do fuzzy_hashes[#fuzzy_hashes + 1] = m.found fuzzy_queried[#fuzzy_queried + 1] = m.queried fuzzy_rules[#fuzzy_rules + 1] = m.rule fuzzy_probs[#fuzzy_probs + 1] = m.prob fuzzy_flags[#fuzzy_flags + 1] = m.flag end table.insert(row, fuzzy_hashes) table.insert(row, fuzzy_queried) table.insert(row, fuzzy_rules) table.insert(row, fuzzy_probs) table.insert(row, fuzzy_flags) -- ASN information local asn, country, ipnet = 0, '--', '--' local pool = task:get_mempool() ret = pool:get_variable("asn") if ret then asn = ret end ret = pool:get_variable("country") if ret then country = ret:sub(1, 2) end ret = pool:get_variable("ipnet") if ret then ipnet = ret end table.insert(row, asn) table.insert(row, country) table.insert(row, ipnet) -- Symbols info if settings.enable_symbols then local symbols = task:get_symbols_all() local syms_tab = {} local scores_tab = {} local options_tab = {} for _, s in ipairs(symbols) do table.insert(syms_tab, s.name or '') table.insert(scores_tab, s.score) if s.options then table.insert(options_tab, table.concat(s.options, ',')) else table.insert(options_tab, ''); end end table.insert(row, syms_tab) table.insert(row, scores_tab) table.insert(row, options_tab) -- Groups data local groups = task:get_groups() local groups_tab = {} local gr_scores_tab = {} for gr, sc in pairs(groups) do table.insert(groups_tab, gr) table.insert(gr_scores_tab, sc) end table.insert(row, groups_tab) table.insert(row, gr_scores_tab) end -- Extra columns if #settings.extra_columns > 0 then for _, col in ipairs(settings.extra_columns) do local elts = col.real_selector(task) if elts then table.insert(row, elts) else table.insert(row, col.default_value) end end end -- Custom data for k, rule in pairs(settings.custom_rules) do if not custom_rows[k] then custom_rows[k] = {} end table.insert(custom_rows[k], lua_clickhouse.row_to_tsv(rule.get_row(task))) end for name, tbl_config in pairs(extra_tables) do if not extra_table_rows[name] then extra_table_rows[name] = {} end local ok, row_data = pcall(tbl_config.get_row, task) if ok and row_data then -- get_row can return a single row or array of rows if row_data[1] and type(row_data[1]) == 'table' then for _, single_row in ipairs(row_data) do local tsv = lua_clickhouse.row_to_tsv(single_row) table.insert(extra_table_rows[name], tsv) used_memory = used_memory + #tsv end else local tsv = lua_clickhouse.row_to_tsv(row_data) table.insert(extra_table_rows[name], tsv) used_memory = used_memory + #tsv end elseif not ok then rspamd_logger.errx(task, 'error in get_row for extra table %s: %s', name, row_data) end end local dominated = false if settings.limits.max_rows > 0 and nrows > settings.limits.max_rows * 10 then dominated = true rspamd_logger.errx(task, 'row count limit exceeded 10x: %d rows (limit %d), discarding data', nrows, settings.limits.max_rows) end if settings.limits.max_memory > 0 and used_memory >= settings.limits.max_memory * 10 then dominated = true rspamd_logger.errx(task, 'memory limit exceeded 10x: %d bytes (limit %d), discarding data', used_memory, settings.limits.max_memory) end if dominated then nrows = 0 used_memory = 0 data_rows = {} custom_rows = {} extra_table_rows = {} collectgarbage() return end local tsv_row = lua_clickhouse.row_to_tsv(row) used_memory = used_memory + #tsv_row data_rows[#data_rows + 1] = tsv_row nrows = nrows + 1 lua_util.debugm(N, task, "add clickhouse row %s / %s; used memory: %s / %s", nrows, settings.limits.max_rows, used_memory, settings.limits.max_memory) end local function do_remove_partition(ev_base, cfg, table_name, partition, method_override) lua_util.debugm(N, rspamd_config, "removing partition %s.%s", table_name, partition) local upstream = settings.upstream:get_upstream_round_robin() if not upstream then rspamd_logger.errx(rspamd_config, "no clickhouse upstream available; cannot remove partition %s.%s", table_name, partition) return end local remove_partition_sql = "ALTER TABLE ${table_name} ${remove_method} PARTITION '${partition}'" local method = method_override or settings.retention.method local remove_method = (method == 'drop') and 'DROP' or 'DETACH' local sql_params = { ['table_name'] = table_name, ['remove_method'] = remove_method, ['partition'] = partition } local sql = lua_util.template(remove_partition_sql, sql_params) local ch_params = { body = sql, ev_base = ev_base, config = cfg, } local err, _ = lua_clickhouse.generic_sync(upstream, settings, ch_params, sql) if err then rspamd_logger.errx(rspamd_config, "cannot detach partition %s:%s from server %s: %s", table_name, partition, settings['server'], err) return end rspamd_logger.infox(rspamd_config, 'detached partition %s:%s on server %s', table_name, partition, settings['server']) end --[[ nil - file is not writable, do not perform removal 0 - it's time to perform removal <int> - how many seconds wait until next run ]] local function get_last_removal_ago() local ts_file = string.format('%s/%s', rspamd_paths['DBDIR'], 'clickhouse_retention_run') local last_ts local current_ts = os.time() local function write_ts_to_file() local write_file, err = io.open(ts_file, 'w') if err then rspamd_logger.errx(rspamd_config, 'Failed to open %s, will not perform retention: %s', ts_file, err) return nil end local res res, err = write_file:write(tostring(current_ts)) if err or res == nil then write_file:close() rspamd_logger.errx(rspamd_config, 'Failed to write %s, will not perform retention: %s', ts_file, err) return nil end write_file:close() return true end local f, err = io.open(ts_file, 'r') if err then lua_util.debugm(N, rspamd_config, 'Failed to open %s: %s', ts_file, err) else last_ts = tonumber(f:read('*number')) f:close() end if last_ts == nil or (last_ts + settings.retention.period) <= current_ts then return write_ts_to_file() and 0 end if last_ts > current_ts then -- Clock skew detected, overwrite last_ts with current_ts and wait for the next -- retention period rspamd_logger.errx(rspamd_config, 'Last collection time is in future: %s; overwrite it with %s in %s', last_ts, current_ts, ts_file) return write_ts_to_file() and -1 end return (last_ts + settings.retention.period) - current_ts end local function clickhouse_maybe_send_data_periodic(cfg, ev_base, now) local need_collect = false local reason if nrows == 0 then lua_util.debugm(N, cfg, "no need to send data, as there are no rows to collect") return settings.check_timeout end if final_call then lua_util.debugm(N, cfg, "no need to send data, final call has been issued") return 0 end if settings.limits.max_rows > 0 then if nrows > settings.limits.max_rows then need_collect = true reason = string.format('limit of rows has been reached: %d', nrows) 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 reason = string.format('limit of time since last collection has been reached: %d seconds passed ' .. '(%d seconds trigger)', math.floor(now - last_collection), math.floor(settings.limits.max_interval)) 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 local saved_rows = data_rows local saved_custom = custom_rows local saved_extra = extra_table_rows nrows = 0 last_collection = now used_memory = 0 data_rows = {} custom_rows = {} extra_table_rows = {} clickhouse_send_data(nil, ev_base, reason, saved_rows, saved_custom, saved_extra) if settings.collect_garbage then collectgarbage() end end return settings.check_timeout end local function clickhouse_remove_old_partitions(cfg, ev_base) local last_time_ago = get_last_removal_ago() if last_time_ago == nil then rspamd_logger.errx(rspamd_config, "Failed to get last run time. Disabling retention") return false elseif last_time_ago ~= 0 then return last_time_ago end local upstream = settings.upstream:get_upstream_round_robin() if not upstream then rspamd_logger.errx(rspamd_config, "no clickhouse upstream available; cannot run retention pass") return false end local partition_to_remove_sql = "SELECT partition, table " .. "FROM system.parts WHERE table IN ('${tables}') " .. "GROUP BY partition, table " .. "HAVING max(max_date) < toDate(now() - interval ${month} month)" local table_names = { 'rspamd' } local table_retention = { rspamd = { period_months = settings.retention.period_months, method = settings.retention.method } } for _, tbl_config in pairs(extra_tables) do if tbl_config.retention and tbl_config.retention.enable then table.insert(table_names, tbl_config.table_name) table_retention[tbl_config.table_name] = { period_months = tbl_config.retention.period_months, method = tbl_config.retention.method } end end -- Use minimum retention period to find all candidate partitions local min_period = settings.retention.period_months for _, ret in pairs(table_retention) do if ret.period_months < min_period then min_period = ret.period_months end end local tables = table.concat(table_names, "', '") local sql_params = { tables = tables, month = min_period, } local sql = lua_util.template(partition_to_remove_sql, sql_params) local ch_params = { ev_base = ev_base, config = cfg, } local err, rows = lua_clickhouse.select_sync(upstream, settings, ch_params, sql) if err then rspamd_logger.errx(rspamd_config, "cannot send data to clickhouse server %s: %s", settings['server'], err) else fun.each(function(row) local ret = table_retention[row.table] if ret then do_remove_partition(ev_base, cfg, row.table, row.partition, ret.method) else do_remove_partition(ev_base, cfg, row.table, row.partition) end end, rows) end -- settings.retention.period is added on initialisation, see below return settings.retention.period end local function upload_clickhouse_schema(upstream, ev_base, cfg, initial) local ch_params = { ev_base = ev_base, config = cfg, } local errored = false -- Upload a single element of the schema local function upload_schema_elt(v) if errored then rspamd_logger.errx(rspamd_config, "cannot upload schema '%s' on clickhouse server %s: due to previous errors", v, upstream:get_addr():to_string(true)) return end local sql = v local err, reply = lua_clickhouse.generic_sync(upstream, settings, ch_params, sql) if err then rspamd_logger.errx(rspamd_config, "cannot upload schema '%s' on clickhouse server %s: %s", sql, upstream:get_addr():to_string(true), err) errored = true return end lua_util.debugm(N, rspamd_config, 'uploaded clickhouse schema element %s to %s: %s', v, upstream:get_addr():to_string(true), reply) end -- Process element and return nil if statement should be skipped local function preprocess_schema_elt(v) if type(v) == 'string' then return lua_util.template(v, { SCHEMA_VERSION = tostring(schema_version) }) elseif type(v) == 'table' then -- Pair of statement + boolean if initial == v[2] then return lua_util.template(v[1], { SCHEMA_VERSION = tostring(schema_version) }) else lua_util.debugm(N, rspamd_config, 'skip clickhouse schema element %s: schema already exists', v) end end return nil end -- Apply schema elements sequentially, users additions are concatenated to the tail fun.each(upload_schema_elt, -- Also template schema version fun.filter(function(v) return v ~= nil end, fun.map(preprocess_schema_elt, fun.chain(clickhouse_schema, settings.schema_additions) ) ) ) end local function maybe_apply_migrations(upstream, ev_base, cfg, version) local ch_params = { ev_base = ev_base, config = cfg, } -- Apply migrations sequentially local function migration_recursor(i) if i < schema_version then if migrations[i] then -- We also need to apply statements sequentially local function sql_recursor(j) if migrations[i][j] then local sql = migrations[i][j] local ret = lua_clickhouse.generic(upstream, settings, ch_params, sql, function(_, _) rspamd_logger.infox(rspamd_config, 'applied migration to version %s from version %s: %s', i + 1, version, sql:gsub('[\n%s]+', ' ')) if j == #migrations[i] then -- Go to the next migration migration_recursor(i + 1) else -- Apply the next statement sql_recursor(j + 1) end end, function(_, err) rspamd_logger.errx(rspamd_config, "cannot apply migration %s: '%s' on clickhouse server %s: %s", i, sql, upstream:get_addr():to_string(true), err) end) if not ret then rspamd_logger.errx(rspamd_config, "cannot apply migration %s: '%s' on clickhouse server %s: cannot make request", i, sql, upstream:get_addr():to_string(true)) end end end sql_recursor(1) else -- Try another migration migration_recursor(i + 1) end end end migration_recursor(version) end local function add_extra_columns(upstream, ev_base, cfg) local ch_params = { ev_base = ev_base, config = cfg, } -- Apply migrations sequentially local function columns_recursor(i) if i <= #settings.extra_columns then local col = settings.extra_columns[i] local prev_column if i == 1 then prev_column = 'MIMERcpt' else prev_column = settings.extra_columns[i - 1].name end local sql = string.format('ALTER TABLE rspamd ADD COLUMN IF NOT EXISTS `%s` %s AFTER `%s`', col.name, col.type, prev_column) if col.comment then sql = sql .. string.format(", COMMENT COLUMN IF EXISTS `%s` '%s'", col.name, col.comment) end local ret = lua_clickhouse.generic(upstream, settings, ch_params, sql, function(_, _) rspamd_logger.infox(rspamd_config, 'added extra column %s (%s) after %s', col.name, col.type, prev_column) -- Apply the next statement columns_recursor(i + 1) end, function(_, err) rspamd_logger.errx(rspamd_config, "cannot apply add column alter %s: '%s' on clickhouse server %s: %s", i, sql, upstream:get_addr():to_string(true), err) end) if not ret then rspamd_logger.errx(rspamd_config, "cannot apply add column alter %s: '%s' on clickhouse server %s: cannot make request", i, sql, upstream:get_addr():to_string(true)) end end end columns_recursor(1) end local function check_rspamd_table(upstream, ev_base, cfg) local ch_params = { ev_base = ev_base, config = cfg, } local sql = [[EXISTS TABLE rspamd]] local err, rows = lua_clickhouse.select_sync(upstream, settings, ch_params, sql) if err then rspamd_logger.errx(rspamd_config, "cannot check rspamd table in clickhouse server %s: %s", upstream:get_addr():to_string(true), err) return end if rows[1] and rows[1].result then if tonumber(rows[1].result) == 1 then -- Apply migration upload_clickhouse_schema(upstream, ev_base, cfg, false) rspamd_logger.infox(rspamd_config, 'table rspamd exists, check if we need to apply migrations') maybe_apply_migrations(upstream, ev_base, cfg, 1) else -- Upload schema rspamd_logger.infox(rspamd_config, 'table rspamd does not exists, upload full schema') upload_clickhouse_schema(upstream, ev_base, cfg, true) end else rspamd_logger.errx(rspamd_config, "unexpected reply on EXISTS command from server %s: %s", upstream:get_addr():to_string(true), rows) end end local function check_clickhouse_upstream(upstream, ev_base, cfg) if not upstream:get_addr() then rspamd_logger.infox(rspamd_config, 'skipping clickhouse upstream %s: address not resolved yet', upstream:get_name()) return end local ch_params = { ev_base = ev_base, config = cfg, } -- If we have some custom rules, we just send its schema to the upstream for k, rule in pairs(settings.custom_rules) do if rule.schema then local sql = lua_util.template(rule.schema, settings) local err, _ = lua_clickhouse.generic_sync(upstream, settings, ch_params, sql) if err then rspamd_logger.errx(rspamd_config, 'cannot send custom schema %s to clickhouse server %s: ' .. 'cannot make request (%s)', k, upstream:get_addr():to_string(true), err) end end end -- Now check the main schema and apply migrations if needed local sql = [[SELECT MAX(Version) as v FROM rspamd_version]] local err, rows = lua_clickhouse.select_sync(upstream, settings, ch_params, sql) if err then if rows and rows.code == 404 then rspamd_logger.infox(rspamd_config, 'table rspamd_version does not exist, check rspamd table') check_rspamd_table(upstream, ev_base, cfg) else rspamd_logger.errx(rspamd_config, "cannot get version on clickhouse server %s: %s", upstream:get_addr():to_string(true), err) end else upload_clickhouse_schema(upstream, ev_base, cfg, false) local version = tonumber(rows[1].v) maybe_apply_migrations(upstream, ev_base, cfg, version) end if #settings.extra_columns > 0 then add_extra_columns(upstream, ev_base, cfg) end end local opts = rspamd_config:get_all_opt('clickhouse') if opts then -- Legacy `limit` options if opts.limit and not opts.limits then settings.limits.max_rows = opts.limit end for k, v in pairs(opts) do if k == 'custom_rules' then if not v[1] then v = { v } end for i, rule in ipairs(v) do if rule.schema and rule.first_row and rule.get_row then local first_row, get_row local loadstring = loadstring or load local ret, res_or_err local chunk, err = loadstring(rule.first_row) if chunk then ret, res_or_err = pcall(chunk) else ret = false res_or_err = err end if not ret or type(res_or_err) ~= 'function' then rspamd_logger.errx(rspamd_config, 'invalid first_row (%s) - must be a function', res_or_err) else first_row = res_or_err end chunk, err = loadstring(rule.get_row) if chunk then ret, res_or_err = pcall(chunk) else ret = false res_or_err = err end if not ret or type(res_or_err) ~= 'function' then rspamd_logger.errx(rspamd_config, 'invalid get_row (%s) - must be a function', res_or_err) else get_row = res_or_err end if first_row and get_row then local name = rule.name or tostring(i) settings.custom_rules[name] = { schema = rule.schema, first_row = first_row, get_row = get_row, } end else rspamd_logger.errx(rspamd_config, 'custom rule has no required attributes: schema, first_row and get_row') end end elseif k == 'limits' or k == 'retention' then for sk, sv in pairs(v) do settings[k][sk] = sv end else settings[k] = lua_util.deepcopy(v) end end if not settings['server'] and not settings['servers'] then rspamd_logger.infox(rspamd_config, 'no servers are specified, disabling module') lua_util.disable_module(N, "config") else local lua_maps = require "lua_maps" settings['from_map'] = lua_maps.map_add('clickhouse', 'from_tables', 'regexp', 'clickhouse specific domains') settings.upstream = upstream_list.create(rspamd_config, settings['server'] or settings['servers'], 8123) if not settings.upstream then rspamd_logger.errx(rspamd_config, 'cannot parse clickhouse address: %s', settings['server'] or settings['servers']) lua_util.disable_module(N, "config") return end if settings.exceptions then local maps_expressions = require "lua_maps_expressions" settings.exceptions = maps_expressions.create(rspamd_config, settings.exceptions, N) end if settings.preset then local preset_cols = extra_column_presets[settings.preset] if not preset_cols then rspamd_logger.errx(rspamd_config, 'clickhouse: unknown preset %s, ignoring', settings.preset) else -- Inject curated extra_columns from the named preset. User-supplied -- entries with the same name take precedence, so we collect existing -- names first. We support both array and map forms of user-supplied -- extra_columns (the subsequent processing loop accepts either). local existing_names = {} if settings.extra_columns then if settings.extra_columns[1] then for _, c in ipairs(settings.extra_columns) do if c.name then existing_names[c.name] = true end end else for k, c in pairs(settings.extra_columns) do existing_names[k] = true if type(c) == 'table' and c.name then existing_names[c.name] = true end end end else settings.extra_columns = {} end local injected = {} local is_array_form = settings.extra_columns[1] ~= nil or next(settings.extra_columns) == nil for _, preset in ipairs(preset_cols) do if not existing_names[preset.name] then -- Deep copy so we do not mutate the module-level preset table local entry = lua_util.deepcopy(preset) if is_array_form then table.insert(settings.extra_columns, entry) else settings.extra_columns[preset.name] = entry end table.insert(injected, preset.name) end end if #injected > 0 then rspamd_logger.infox(rspamd_config, 'clickhouse: preset %s injected extra_columns: %s', settings.preset, table.concat(injected, ', ')) end end end if settings.extra_columns then -- Check sanity and create selector closures local lua_selectors = require "lua_selectors" local columns_transformed = {} local need_sort = false -- Select traverse function depending on what we have local iter_func = settings.extra_columns[1] and ipairs or pairs for col_key, col_data in iter_func(settings.extra_columns) do -- Array based extra columns local col_name = col_data.name or col_key if not col_data.selector or not col_data.type then rspamd_logger.errx(rspamd_config, 'cannot add clickhouse extra row %s: no type or no selector', col_name) else local is_array = false if col_data.type:lower():match('^array') then is_array = true end local selector = lua_selectors.create_selector_closure(rspamd_config, col_data.selector, col_data.delimiter or '', is_array) if not selector then rspamd_logger.errx(rspamd_config, 'cannot add clickhouse extra row %s: bad selector: %s', col_name, col_data.selector) else if not col_data.default_value then if is_array then col_data.default_value = {} else col_data.default_value = '' end end col_data.real_selector = selector if not col_data.name then col_data.name = col_name need_sort = true end table.insert(columns_transformed, col_data) end end end -- Convert extra columns from a map to an array sorted by column name to -- preserve strict order when doing altering if need_sort then rspamd_logger.infox(rspamd_config, 'sort extra columns as they are not configured as an array') table.sort(columns_transformed, function(c1, c2) return c1.name < c2.name end) end settings.extra_columns = columns_transformed end rspamd_config:register_symbol({ name = 'CLICKHOUSE_COLLECT', type = 'idempotent', callback = clickhouse_collect, flags = 'empty,explicit_disable,ignore_passthrough', augmentations = { string.format("timeout=%f", settings.timeout) }, }) rspamd_config:register_finish_script(function(task) if nrows > 0 then final_call = true local saved_rows = data_rows local saved_custom = custom_rows local saved_extra = extra_table_rows nrows = 0 data_rows = {} used_memory = 0 custom_rows = {} extra_table_rows = {} clickhouse_send_data(task, nil, 'final collection', saved_rows, saved_custom, saved_extra) if settings.collect_garbage then collectgarbage() end end end) -- Create tables on load rspamd_config:add_on_load(function(cfg, ev_base, worker) if worker:is_scanner() then rspamd_config:add_periodic(ev_base, 0, clickhouse_maybe_send_data_periodic, true) end if worker:is_primary_controller() then local upstreams = settings.upstream:all_upstreams() for _, up in ipairs(upstreams) do check_clickhouse_upstream(up, ev_base, cfg) end if settings.retention.enable and settings.retention.method ~= 'drop' and settings.retention.method ~= 'detach' then rspamd_logger.errx(rspamd_config, "retention.method should be either 'drop' or 'detach' (now: %s). Disabling retention", settings.retention.method) settings.retention.enable = false end if settings.retention.enable and settings.retention.period_months < 1 or settings.retention.period_months > 1000 then rspamd_logger.errx(rspamd_config, "please, set retention.period_months between 1 and 1000 (now: %s). Disabling retention", settings.retention.period_months) settings.retention.enable = false end local period = lua_util.parse_time_interval(settings.retention.run_every) if settings.retention.enable and period == nil then rspamd_logger.errx(rspamd_config, "invalid value for retention.run_every (%s). Disabling retention", settings.retention.run_every) settings.retention.enable = false end if settings.retention.enable then settings.retention.period = period rspamd_logger.infox(rspamd_config, "retention will be performed each %s seconds for %s month with method %s", period, settings.retention.period_months, settings.retention.method) rspamd_config:add_periodic(ev_base, 0, clickhouse_remove_old_partitions, false) end end end) -- Expose API for other plugins to register extra tables rspamd_plugins['clickhouse'] = { register_extra_table = register_extra_table, unregister_extra_table = unregister_extra_table, get_extra_tables = get_extra_tables, is_enabled = function() return settings.upstream ~= nil end, } end end
Save File
Cancel