Logstash Issue - export kafka messages into tsv format

[aniketdatir (aniket)]

Hi Elastic team,

I am trying to export Kafka messages into tab separated file (tsv) file using below logstash configuration file......But not able to get desired output.

input {
kafka {
bootstrap_servers => "host:port"
topics => ["kafka_topic"]
max_poll_records =>500
}
}
output {
csv {
fields => ["message","@timestamp"]
csv_options => {"write_headers" => "true" "headers" =>["message","@timestamp"] "col_sep" => "\t"}
path => "/tmp/abc.tsv"
}
}

Below are the problems facing using above conf file

1. not able to export in tsv file.' \t' is appending before data in file.
   for ex. a\tb instead a b.
2. headers are repeating after every row
   for ex. a\tb
   1\t2
   a\tb
   3\t4
3. why these many opening and closing happened while running this configuration

[INFO ] 2022-07-11 13:28:08.522 [[main]>worker2] csv - Opening file {:path=>"/tmp/abc.tsv"}
[INFO ] 2022-07-11 13:28:47.067 [[main]>worker1] csv - Closing file /tmp/abc.tsv
[INFO ] 2022-07-11 13:29:27.352 [[main]>worker0] csv - Opening file {:path=>"/tmp/abc.tsv"}
[INFO ] 2022-07-11 13:29:42.042 [[main]>worker0] csv - Closing file /tmp/abc.tsv

desired output is
for ex. a b
1 2
3 4

Kindly let me know what I am missing & correct me
Thanks

[Badger]

1. Ruby code (at least in some places) would substitute the \t with a tab character. The logstash configuration compiler does not do that by default -- you need a literal tab in your configuration. There is a 7 year old open issue on this. Enabling the config.support_escapes option in logstash.yml may fix this.

2. Yes, headers are repeated. There is a 6 year old open issue on this.

3. Well, this one I can help on. The csv output borrows the close_stale_files method from the file output, and the file output has option to control how often this is called. This is a minimum time, it may take more than 60 seconds for the file to be closed.

   stale_cleanup_interval => 60