fluent.conf

<source>
  @type tail
  tag postgres.postgres_csv
  path /home/postgres/pgdata/pgroot/pg_log/*.csv
  read_from_head true
  follow_inodes true
  pos_file /home/postgres/pgdata/tmp/fluentd.csv.pos
  pos_file_compaction_interval 72h
  multiline_flush_interval 5s

  <parse>
    @type multiline_csv
    format_firstline /^\d{4}-\d{1,2}-\d{1,2} \d{2}:\d{2}:\d{2}/
    keys time,user_name,database_name,process_id,connection_from,session_id,session_line_num,command_tag,session_start_time,virtual_transaction_id,transaction_id,error_severity,sql_state_code,message,detail,hint,internal_query,internal_query_pos,context,query,query_pos,location,application_name
    types time:time,process_id:integer,session_line_num:integer,session_start_time:time,transaction_id:integer,internal_query_pos:integer,query_pos:integer
    time_key time
    time_format %Y-%m-%d %H:%M:%S.%N %Z
    keep_time_key true
    parser_type fast
  </parse>
</source>

<filter postgres.*>
  @type record_transformer
  <record>
    host "#{ENV['NODE_NAME']}"
    tag
  </record>
</filter>

<filter postgres.postgres_csv>
  @type postgresql_slowlog
  output_key query
</filter>

<filter postgres.postgres_csv>
  @type postgresql_redactor
  input_key query
  output_key sql
  fingerprint_key fingerprint
</filter>

<filter postgres.postgres_csv>
  @type marginalia
  key sql
  strip_comment true
</filter>

<match postgres.*>
  @type stdout
  time_key time
  <format>
    @type json
    time_type float
  </format>
  <inject>
    time_type string
    time_format %Y-%m-%dT%H:%M:%S.%L
    time_key time
  </inject>
</match>