Parsing CSV with multi line (postgres csvlog)

[masaiful]

Hi,

I have postgres log using csvlog such as:

2022-12-14 09:48:53.026 CET,"myuser","postgres",270752,"[local]",63998df5.421a0,2,"authentication",2022-12-14 09:48:53 CET,10/2532,0,LOG,00000,"connection authorized: user=myuser database=postgres application_name=psql",,,,,,,,"PerformAuthentication, postinit.c:292","","client backend"
2022-12-15 04:41:25.924 CET,"writeuser","mydb",128237,"10.4.3.9:55578",639a918c.1f4ed,3,"idle",2022-12-15 04:16:28 CET,,0,LOG,00000,"disconnection: session time: 0:24:57.588 user=writeuser database=mydb host=10.4.3.9 port=55578",,,,,,,,"log_disconnections, postgres.c:4767","master_conn","client backend"
2022-12-15 04:41:20.727 CET,"writeuser","mydb",243567,"10.2.3.2:42096",639a938d.3b76f,16,"INSERT",2022-12-15 04:25:01 CET,94/42049981,1759425267,ERROR,23505,"duplicate key value violates unique constraint ""col_unique""","Key (col1)=(\x9busbdhd) already exists.",,,,,"WITH par AS (INSERT INTO table (col1
			VALUES ($1)
			ON CONFLICT (col2 DO UPDATE SET
				col2 = EXCLUDED.col2
			RETURNING col2
		)


	insert into tab2
	(col1,col2,col3,col4,col5,col6,col7,alias1,alias2,alias3,alias4)
	select 1 as col1, $2,$3,$4,$5,$6,

			(SELECT col FROM par),


	        (select new.col4 from (values

		          ($7)
		         ,
		          ($8)
		         ,
		          ($9)
		         ,
		          ($10)
		         ,
		          ($11)

			   ) as new(col4) left join tab2 a on new.col4 = a.col3 where a.col4 isnull limit 1) as col5,

	        	(select new.col2 from (values

		          ($12)
		         ,
		          ($13)
		         ,
		          ($14)
		         ,
		          ($15)
		         ,
		          ($16)

			   ) as new(col3) left join tab1 a on new.col2 = a.col3 where a.col4 isnull limit 1)

	          as alias2,


	        NULL

	        as alias3,


	        NULL

	         as alias4
			",,"_bt_check_unique, nbtinsert.c:656","master_conn","client backend"
2022-12-15 04:41:26.050 CET,"writeuser","mydb",130056,"10.40.31.116:50652",639a92bd.1fc08,392,"SELECT",2022-12-15 04:21:33 CET,679/227106160,0,LOG,00000,"duration: 3442.218 ms  execute <unnamed>: SELECT col1 FROM
      newtable
     WHERE col1=$1 AND col2=$2
        group by col2","parameters: $1 = '4', $2 = 'GB'",,,,,,,"exec_execute_message, postgres.c:2265","master_conne","client backend"
2022-12-09 14:15:13.093 CET,"userZ","dbA",231892,"10.1.3.15:55124",639326d5.389d4,11,"SELECT",2022-12-09 13:15:17 CET,7/0,0,LOG,X0002,"duration: 1661.790 ms",,,,,,,,"exec_simple_query, postgres.c:1324","psql","client backend"

I tried parsing this with parse_csv, but multi line doesn't work with this configuration, I believe it is stated parse_csv only supports one liner:

[sources.postgres_csv]
type = "file"
include = [ "postgresql*.csv" ]
read_from = "beginning"

[transforms.pg_parsed]
type = "remap"
inputs = [ "postgres_csv" ]
source = """
  line = parse_csv!(.message)
  .timestamp = line[0]
  .username = line[1]
  .dbname = line[2]
  .pid = line[3]
  .client_addr = line[4]
  .session_id = line[5]
  .session_line_num = line[6]
  .command_tag = line[7]
  .session_start_time = line[8]
  .virtual_transaction_id = line[9]
  .transaction_id = line[10]
  .severity = line[11]
  .sqlcode = line[12]
  .message = line[13]
  .detail = line[14]
  .hint = line[15]
  .internal_query = line[16]
  .internal_query_pos = line[17]
  .context = line[18]
  .query = line[19]
  .query_pos = line[20]
  .location = line[21]
  .application_name = line[22]
"""

I also tried following this page https://vector.dev/guides/advanced/parsing-csv-logs-with-lua/ and https://vector.dev/guides/advanced/merge-multiline-logs-with-lua/ with no luck. I keep getting this error, my suspicion is with column naming(?):

ERROR transform{component_kind="transform" component_id=pg_lua component_type=lua component_name=pg_lua}: 
vector::internal_events::lua: Error in building lua script. error=RuntimeErrorHooksProcess { source: RuntimeError("[string \"src/transforms/lua/v2/mod.rs:196:9\"]:7: 
table index is nil\nstack traceback:\n\t[C]: 
in metamethod 'newindex'\n\t[string \"src/transforms/lua/v2/mod.rs:196:9\"]:7: in function <[string \"src/transforms/lua/v2/mod.rs:196:9\"]:1>") }
error_type="script_failed" error_code="runtime_error_hook_process" stage="processing" internal_log_rate_limit=true

If anyone can help or has an example I can follow, I would really appreciate it.
Thank you.

PS: I am not a programmer

[jszwedko]

Hey @masaiful !

Yeah, I see. I think what you'll need to do is make sure that the multiline logs are parsed as a single event before parsing them as CSV using parse_csv. You can make use of the https://vector.dev/docs/reference/configuration/transforms/reduce/ transform to reduce multi-line logs into one line. Some sources, like the file source, also support multiline configuration to reduce the lines during ingest.

[masaiful]

Thanks, @jszwedko! I did tried with multiline in file source type earlier with no luck. For whatever reasons, it doesn't transform as I expected. Not sure what went wrong in that part.

But, I managed to make it work with reduce and then remap. The only thing that I can't seem to find a solution is how do I reorder the output, from A-Z to something I am familiar with the log format. I am planning to use elasticsearch and/or log_to_metrics for sinks, not sure if reordering is necessary.