How to pass multiple csv files with different names and different content?

Elastic Stack Logstash
1

Hello,

I want to pass into Elastisearch several CSV files with a different format each other. Till now, I have created a Logstash conf file that handles only one CSV with a specific name under a specific path. But when I am using a * to catch all the files under the directory the result is not what I want. Below you can find my Logstash conf file and the CSV files. Could anyone give some advice about what I am doing wrong? Thanks in advance.

input {
file {
path =>"/home/kibana/Downloads/scde1_report_MTSMS_2019-02-05.csv"
start_position => "beginning"
sincedb_path => "/dev/null"
}

	file {
            path =>"/home/kibana/Downloads/scde1_report_MTSMS_Throttling2019-01-23.csv"
            start_position => "beginning"
            sincedb_path => "/dev/null"
    }

}
filter {
if [path] == "/home/kibana/Downloads/scde1_report_MTSMS_2019-02-05.csv"
{
if ([message] =~ "\bDate\b") {
drop { }
} else {
csv {
separator => ","
columns => [ "Date","Hour","Avg_TPS","Avg_Top_1%_TPS","Avg_Top_10%_TPS","Avg_Proc","Avg_Top_1%_Proc","Avg_Top_10%_Proc" ]

    }	}
	}
	else if [path] == "/home/kibana/Downloads/scde1_report_MTSMS_Throttling2019-01-23.csv"
	{
	 if ([message] =~ "\bDate\b") {
    drop { }
} else {
    csv {
            separator => ","
            columns => [ "Date","Hour","Avg_TPS","Avg_Top_1%_TPS","Avg_Top_10%_TPS" ]

    }	}

}
mutate{

            add_field => {
                    "DateHour" => "%{Date} %{Hour}"
                    }
            convert => {
                    "Avg_TPS" => "float"
                    "Avg_Top_1%_TPS" => "float"
                    "Avg_Top_10%_TPS" => "float"
                    "Avg_Proc" => "float"
                    "Avg_Top_1%_Proc" => "float"
                    "Avg_Top_10%_Proc" => "float"
                    }
            }

date {
match => ["DateHour","YYYY-MM-dd HH ","ISO8601"]
timezone => "Europe/Athens"
target => "DateHour"
remove_field => ["message","path","host","Date","Hour"]
}

}
output {
elasticsearch{
hosts => "localhost:9200"
index => "report"
document_type => "MT_SMS"
}
stdout{}

}

Config for multiple files

input {
file {
path =>"/home/kibana/Downloads/*.csv"
start_position => "beginning"
sincedb_path => "/dev/null"
}

	file {
            path =>"/home/kibana/Downloads/*.csv"
            start_position => "beginning"
            sincedb_path => "/dev/null"
    }

}
filter {
if [path] == "/home/kibana/Downloads/*.csv"
{
if ([message] =~ "\bDate\b") {
drop { }
} else {
csv {
separator => ","
columns => [ "Date","Hour","Avg_TPS","Avg_Top_1%_TPS","Avg_Top_10%_TPS","Avg_Proc","Avg_Top_1%_Proc","Avg_Top_10%_Proc" ]

    }	}
	}
	else if [path] == "/home/kibana/Downloads/*.csv"
	{
	 if ([message] =~ "\bDate\b") {
    drop { }
} else {
    csv {
            separator => ","
            columns => [ "Date","Hour","Avg_TPS","Avg_Top_1%_TPS","Avg_Top_10%_TPS" ]

    }	}

}
mutate{

            add_field => {
                    "DateHour" => "%{Date} %{Hour}"
                    }
            convert => {
                    "Avg_TPS" => "float"
                    "Avg_Top_1%_TPS" => "float"
                    "Avg_Top_10%_TPS" => "float"
                    "Avg_Proc" => "float"
                    "Avg_Top_1%_Proc" => "float"
                    "Avg_Top_10%_Proc" => "float"
                    }
            }

date {
match => ["DateHour","YYYY-MM-dd HH ","ISO8601"]
timezone => "Europe/Athens"
target => "DateHour"
remove_field => ["message","path","host","Date","Hour"]
}

}
output {
elasticsearch{
hosts => "localhost:9200"
index => "report"
document_type => "MT_SMS"
}
stdout{}

}

2

If I am reading it correctly you have

It is never going to go through the second branch.

3

In the first case it is working because under /Downloads/ there is a specific name?

4

Yes.

5

Can you propose me any solution?

6

You need to have a more specific pattern that only matches the files that have the format that the branch parses. If just the date changes then maybe something like

if [path] =~ /scde1_report_MTSMS_Throttling.*\.csv/ {
    [...]
else if [path] =~ /scde1_report_MTSMS.*\.csv/ {
7

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.