fields aggregate during db insertion

[adelbordbari]

Hello! i'm facing an issue with my incoming syslog have this fluent.conf:

<source>
  @type syslog
  port 514
  bind 0.0.0.0
  tag s_514
  source_address_key sensor
  <transport udp>
  </transport>
  message_format rfc5424
</source>

# #---------------------Start firewall
<filter s_514.local0.*>
  @type parser
  key_name message
  reserve_data true
  remove_key_name_field true
  <parse>
    @type csv
    keys rule_id,rule_number,sub_rule_number,ruleset,p1,interface,reason,action,direction,ip_version,p2,p3,p4,p5,p6,p7,p8,tcp_udp,p9,source_ip,dest_ip,src_port,dest_port,p10,p11,p12,p13,p14,p15,p16
  </parse>
</filter>

<match s_514.local0.*>
  @type copy
  <store>
  @type clickhousejson
    host clickhouse
    port 8123
    database default
    table firewall
    distribute false
    buffer_type memory
    buffer_queue_limit 512
    flush_at_shutdown true
    flush_interval 5s
  </store>
  <store>
    @type redis
    host redis
    ttl 60
    <buffer>
      flush_interval 1s
      compress gzip
    </buffer>
  </store>
</match>
# #---------------------END firewall

# #---------------------Start IDS
<filter s_514.local5.*>
  @type parser
  key_name message
  format json
  reserve_data false
</filter>

<filter s_514.local5.*>
  @type record_transformer
  renew_record true
  keep_keys timestamp, flow_id, interface, event_type, src_ip, src_port, dest_ip, dest_port, proto, action, signature_id, signature, severity
  enable_ruby true
  <record>
    timestamp ${record["timestamp"]}
    interface ${record["in_iface"]}
    action ${record["alert"]["action"]}
    signature_id ${record["alert"]["signature_id"]}
    signature ${record["alert"]["signature"]}
    severity ${record["alert"]["severity"]}
  </record>
  remove_keys message
</filter>

<match s_514.local5.*>
  @type clickhousejson
  host clickhouse
  port 8123
  database default
  table ids
  distribute false
  buffer_type memory
  buffer_queue_limit 512
  flush_at_shutdown true
  flush_interval 5s
</match>

i basically have a stream of logs incoming on 514 udp port with syslog. i want to parse two separate facilities differently, and insert the results into two different tables.

i have different sets of fields for each log (since they're, different!)
i check the logs with the code below, and they're fine, as expected

<filter *.local5.*>
    @type stdout
</filter>

my problem is that after insertion into db, both tables have ALL of the fields! with null values. for example i have p2,p3,p4,p5,p6,p7,p8 fields only for my firewall (local0 facility) logs but my IDS table also has them in db.
i run fluentd and clickhouse (db) as docker-compose services. i also use a custom output plugin (that used to work fine, as claimed by the previous dev)

this is my output plugin just in case:

require 'fluent/output'
require 'fluent/config/error'
require 'net/http'
require 'date'
require 'yajl'
require 'json'
require 'time'
require 'securerandom'


module Fluent
    class ClickhouseOutputJSON < BufferedOutput
        Fluent::Plugin.register_output("clickhousejson", self)

        DEFAULT_TIMEKEY = 60 * 60 * 24

        desc "IP or fqdn of ClickHouse node"
        config_param :host, :string, default: "localhost"
        desc "Port of ClickHouse HTTP interface"
        config_param :port, :integer, default: 8123
        desc "Database to use"
        config_param :database, :string, default: "default"
        desc "Table to use"
        config_param :table, :string
        desc "User of Clickhouse database"
        config_param :user, :string, default: "default"
        desc "Password of Clickhouse database"
        config_param :password, :string, default: ""
        desc "Offset in minutes, could be useful to substract timestamps because of timezones"
        config_param :tz_offset, :integer, default: 0
        desc "Name of internal fluentd time field (if need to use)"
        config_param :datetime_name, :string, default: nil
        config_section :buffer do
            config_set_default :@type, "file"
            config_set_default :path, "/var/log/td-agent/buffer/clickhousejson"
            config_set_default :chunk_keys, ["time"]
            config_set_default :flush_at_shutdown, true
            config_set_default :timekey, DEFAULT_TIMEKEY
        end

        def configure(conf)
            super
            @uri, @uri_params = make_uri(conf)
            @table            = conf["table"]
            @tz_offset        = conf["tz_offset"].to_i
            @datetime_name    = conf["datetime_name"] || "DateTime"
			@is_dist    = conf["distribute"] || "false"
            test_connection(conf)
        end

        def multi_workers_ready?
            true
        end

        def test_connection(conf)
            uri = @uri.clone
            uri.query = URI.encode_www_form(@uri_params.merge({"query" => "SHOW TABLES"}))
            begin
                res = Net::HTTP.get_response(uri)
            rescue Errno::ECONNREFUSED
                raise Fluent::ConfigError, "Couldn't connect to ClickHouse at #{ @uri } - connection refused"
            end
            if res.code != "200"
                raise Fluent::ConfigError, "ClickHouse server responded non-200 code: #{ res.body }"
            end
        end

        def make_uri(conf)
            uri = URI("http://#{ conf["host"] || 'localhost' }:#{ conf["port"] || 8123 }/")
            params = {
                "database" => conf["database"] || "default",
                "user"     => conf["user"] || "default",
                "password" => conf["password"] || "",
                "input_format_skip_unknown_fields" => 0
            }
            return uri, params
        end
		
		$keyQ=[]
		
        def format(tag, timestamp, record)
			
			 #record[@datetime_name] = (timestamp.to_i+Time.zone_offset('+0330'))
			 record[@datetime_name] = timestamp
			 #record["uid"] = SecureRandom.uuid
			 replacements = {
                "(" => "_", ")" => "_", " " => "_","," => "_", "\\" => "/",  "-" => "_",
                "<" => "_", ">" => "_", 
              }
              new_h = Hash[record.map { |k, v| [k.gsub(Regexp.union(replacements.keys), replacements), v.to_s] }]
	  
			 key=[]
			 key << new_h.keys
			  
			 key[0].map { |s| $keyQ.push(s) }
			 $keyQ=$keyQ.uniq
			  
            return Yajl.dump(new_h) + "\n"
        end
		
		def runQuary(quary)
			uri = @uri.clone
			query = {"query" => "#{quary}"}
			uri.query = URI.encode_www_form(@uri_params.merge(query))
			req = Net::HTTP::Post.new(uri)
			http = Net::HTTP.new(uri.hostname, uri.port)
			resp = http.request(req)
			return resp
		end
		
        def write(chunk)
		
#####################################################################################		
			if @is_dist == "false" 
				uri = @uri.clone
				query = {"query" => "INSERT INTO #{@table} FORMAT JSONEachRow"}
				uri.query = URI.encode_www_form(@uri_params.merge(query))
				req = Net::HTTP::Post.new(uri)
				req.body = chunk.read
				rows=req.body.split("\n")
				
				http = Net::HTTP.new(uri.hostname, uri.port)
				resp = http.request(req)
				if resp.code == "200"
					log.info "[#{@table}] IMPORT [ #{rows.count} ] ROWS IN DATABASE #{@database}  TABLE #{(@table).ljust(10)} resCode:#{resp.code} "
				else 
					
					if resp.code == "404"
						
						
						#quary="CREATE TABLE IF NOT EXISTS #{@table}  ( Date Date MATERIALIZED toDate(DateTime), DateTime DateTime) ENGINE = MergeTree(Date, DateTime, 8192)"
						#quary="CREATE TABLE IF NOT EXISTS #{@table}  ( `id` UInt64 ) ENGINE = MergeTree()"
						#quary="CREATE TABLE IF NOT EXISTS #{@table} (DateTime DateTime , uid String) ENGINE = MergeTree() PARTITION BY DateTime ORDER BY (DateTime,uid) ;"
						quary="CREATE TABLE IF NOT EXISTS #{@table} (DateTime DateTime , uid String) ENGINE =MergeTree() ORDER BY toYYYYMM(DateTime);"
						resp = runQuary(quary)
						log.error "[#{@table}] CLCIK responded: #{@table} not exists and create table ; #{resp.code} "
						if resp.code == "200"
							log.info "[#{@table}] CRATE TABLE IF NOT EXIST #{@table.ljust(15)} resCode:#{resp.code} "
						else 
							log.warn "[#{@table}] Clickhouse responded: #{resp.code} ; CLICK_ERROR : #{resp.body}"
						end
					end
					uri = @uri.clone
					query = {"query" => "select name from system.columns WHERE table == '#{@table}'"}
					uri.query = URI.encode_www_form(@uri_params.merge(query))
					req = Net::HTTP::Post.new(uri)
					http = Net::HTTP.new(uri.hostname, uri.port)
					resp = http.request(req)
				
					columns=[]
					columns << resp.body
					columns=columns[0].split("\n")
					key=$keyQ-columns
				
					key.map {|s| 
					uri = @uri.clone
					query = {"query" => "ALTER TABLE  #{@table}  ADD COLUMN IF NOT EXISTS #{s} Nullable(String)"}
					
					log.info "[#{@table}] ADD #{(" COLUMN "+s).ljust(40)}resCode:#{resp.code}"
					uri.query = URI.encode_www_form(@uri_params.merge(query))
					req = Net::HTTP::Post.new(uri)
					http = Net::HTTP.new(uri.hostname, uri.port)
					resp = http.request(req)
					}
				 
					uri = @uri.clone
					query = {"query" => "INSERT INTO #{@table} FORMAT JSONEachRow"}
					uri.query = URI.encode_www_form(@uri_params.merge(query))
					req = Net::HTTP::Post.new(uri)
					req.body = chunk.read
					rows=req.body.split("\n")
					http = Net::HTTP.new(uri.hostname, uri.port)
					resp = http.request(req)
					
					if resp.code == "200"
						log.info "[#{@table}] IMPORT [ #{rows.count} ] ROWS IN DATABASE #{@database}  TABLE #{(@table).ljust(10)} resCode:#{resp.code} "
					end
				end
				#if is_dist true
            end
###########################################################################################	
			
				#if is_dist true
        end
    end


end

this is an example of what i see in stdout (which is what i expect):

2024-11-16 07:58:27.398044453 +0000 s_514.local0.info: {"host":"xx.xx.xx","ident":"filterlog","pid":"58861","msgid":"-","extradata":"[meta sequenceId=\"6000\"]","sensor":"192.168.20.1","rule_id":"1060<1>","rule_number":"166","sub_rule_number":null,"ruleset":null,"p1":"0","interface":"em1_vlan105","reason":"match","action":"pass","direction":"out","ip_version":"4","p2":"0x0","p3":null,"p4":"63","p5":"54524","p6":"0","p7":"none","p8":"17","tcp_udp":"udp","p9":"56","source_ip":"192.168.254.2","dest_ip":"192.168.40.2","src_port":"64863","dest_port":"53","p10":"36","p11":null,"p12":null,"p13":null,"p14":null,"p15":null,"p16":null}

2024-11-16 07:58:27.410763609 +0000 s_514.local5.info: {"timestamp":"2024-11-16T11:28:27.411581+0330","flow_id":923303644754837,"event_type":"alert","src_ip":"192.168.xx.xx","src_port":54069,"dest_ip":"192.168.20.1","dest_port":53,"proto":"UDP","interface":"xx","action":"allowed","signature_id":4294967268,"signature":"DNS","severity":3}