Hello!
I am new to ELK stack and I'm looking for some advice. I have 3 different types of csv's files which are generated every day, I'm sending them to IP address where logstash is configured, than i want to visualize data from those files in kibana. When i was trying things locally i was using file input, where every line from files was treated as new event and than i was using "if else" to apply proper csv filter to file based on number of columns and everything worked as i wanted. Now when i send those files to IP address, http input process files as one event and nothing works. I tried to use split filter but couldn't get it to work 
Should i tried different approach, or I am forgetting about something obvious? here is my config file( i left split filter empty now as I could not get it to work but i suspect i need to use it):
input {
http{
}
}
filter{
split{}
ruby { code => 'event.set("[@metadata][columns]", 1 + event.get("message").count(","))' }
if [@metadata][columns] == 7{
csv{
separator => ","
columns => ["Class","Asset Name","Issue","Value","Severity","Path", "Date"]
}
mutate{
add_tag => "assets"
}
}else if [@metadata][columns] == 8{
csv {
separator => ","
columns => ["Time (ms)","Frame (ms)","GT (ms)","RT (ms)","GPU (ms)","DynRes","Context","Date"]
}
mutate{
add_tag => "fps_profiling"
convert => {
"Time (ms)" => "float"
"Frame (ms)" => "float"
"GT (ms)" => "float"
"RT (ms)" => "float"
"GPU (ms)" => "float"
"DynRes" => "float"
"Context" => "integer"
}
}
} else if [@metadata][columns] == 9{
csv {
separator => ","
columns => ["Percentile","Frame (ms)","GT (ms)","RT (ms)","GPU (ms)","DynRes","Context","Date"]
}
mutate{
add_tag => "fps_profiling"
convert => {
"Percentile" => "float"
"Frame (ms)" => "float"
"GT (ms)" => "float"
"GPU (ms)" => "float"
"DynRes" => "float"
"Context" => "integer"
}
}
}
date {
match => [ "Date", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ"]
target => "Date"
}
}
output {
stdout {codec => rubydebug}
if "fps_profiling" in [tags]{
elasticsearch{
hosts => ["localhost:9200"]
index => "performance_tests"
}
}else {
elasticsearch{
hosts => ["localhost:9200"]
index => "assets_validation"
}
}
}