Logstash conditional in output

Elastic Stack Logstash
1

Hi,
I wanted to know if it is possible to carry out an action based on a conditional in the output section of my logstash configuration file. I am parsing a file into CSV format and want to add a header. I do this by adding the following lines to the csv output plugin:

csv_options => { #specifying the headers for the fields
"write_headers" => true
"headers" => ["prodId", "vendorName", "vendorId", "model", "categoryName", "fob_city", "fob_state", "fob_zip", "spec", "listPrice", "netPrice", "numPerCase", "sellingUnit", "shippingCube", "shippingWeight", "freightClass", "height", "width", "depth", "catalogItem", "multiplier", "hideModel", "callForPricing", "pictureId", "cutSheetId", "hasConfig", "configAccessories", "values_property", "values_value", "certifications_text", "utils_electrical_order", "utils_electrical_amps", "utils_electrical_kw", "utils_electrical_hp", "utils_electrical_mca", "utils_electrical_mocp", "utils_electrical_voltage", "utils_electrical_cycle", "utils_electrical_nema", "utils_electrical_remarks", "utils_electrical_phase", "utils_electrical_electricalConnectionHeight", "utils_electrical_connectionType", "changeState"]
"col_sep" => ","
}
However, since my input file consists of multiple lines, this header is added for ever line that is parsed. In order to prevent this, I want to perform a check of whether the output file path already exists. If it already exists then the header will not be added. Otherwise, header will be added as above.

2

I wanted to know if it is possible to carry out an action based on a conditional in the output section of my logstash configuration file.

Yes...

In order to prevent this, I want to perform a check of whether the output file path already exists.

...but only based on the fields in the current event. You can't check for file existence. I suppose you could do that with a ruby filter that tags each event with a field that indicates whether the file existed when that event was processed. Doing so won't eliminate the problem entirely though.

3

Thank you for your response. Just to make sure, the entire config file is executed for every line of the input file (json format), correct? So if my csv plugin outputs the output to a specific file path, the file will be created after the very first line of the input file is parsed, correct?

4

I added the following snippet to my filter section:

ruby {
code => 'if (file.exist?("C:\Users\prati\Desktop\CE_Training\csvoutput-json-csv-conv.csv")) {
tags => ["headers_created"]
}'
}

And, I added the following condition to my csv_options in the output section:

if ("headers_created" in [tags]) {
csv_options => { #specifying the headers for the fields
"write_headers" => true
"headers" => ["prodId", "vendorName", "vendorId", "model", "categoryName", "fob_city", "fob_state", "fob_zip", "spec", "listPrice", "netPrice", "numPerCase", "sellingUnit", "shippingCube", "shippingWeight", "freightClass", "height", "width", "depth", "catalogItem", "multiplier", "hideModel", "callForPricing", "pictureId", "cutSheetId", "hasConfig", "configAccessories", "values_property", "values_value", "certifications_text", "utils_electrical_order", "utils_electrical_amps", "utils_electrical_kw", "utils_electrical_hp", "utils_electrical_mca", "utils_electrical_mocp", "utils_electrical_voltage", "utils_electrical_cycle", "utils_electrical_nema", "utils_electrical_remarks", "utils_electrical_phase", "utils_electrical_electricalConnectionHeight", "utils_electrical_connectionType", "changeState"]
"col_sep" => ","
}
}

After starting the pipeline in Windows Powershell, I get the following output/error message:

Sending Logstash's logs to C:/Users/prati/Desktop/ELK/logstash/logs which is now configured via log4j2.properties
[2017-09-27T13:22:53,449][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"C:/Users/prati/Desktop/ELK/logstash/modules/fb_apache/configuration"}
[2017-09-27T13:22:53,456][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"C:/Users/prati/Desktop/ELK/logstash/modules/netflow/configuration"}
[2017-09-27T13:22:53,739][ERROR][logstash.agent ] Cannot create pipeline {:reason=>"Expected one of #, => at line 129, column 6 (byte 4269) after output {\n\tcsv { #specifying the fields to split the j
son input into\n\t\tfields => ["prodId", "vendorName", "vendorId", "model", "categoryName", "fob_city", "fob_state", "fob_zip", "spec", "listPrice", "netPrice", "numPerCase", "sellingUni
t", "shippingCube", "shippingWeight", "freightClass", "height", "width", "depth", "catalogItem", "multiplier", "hideModel", "callForPricing", "pictureId", "cutSheetId", "hasConfig", "
configAccessories", "values_property", "values_value", "certifications_text", "utils_electrical_order", "utils_electrical_amps", "utils_electrical_kw", "utils_electrical_hp", "utils_electrical_mc
a", "utils_electrical_mocp", "utils_electrical_voltage", "utils_electrical_cycle", "utils_electrical_nema", "utils_electrical_remarks", "utils_electrical_phase", "utils_electrical_electricalConnect
ionHeight", "utils_electrical_connectionType", "changeState"]\n\t\tpath => ["C:\Users\prati\Desktop\CE_Training\csvoutput-json-csv-conv.csv"]\n\t\t\n\t\t#if (File.size("C:\Users\prati\Desktop\CE
_Training\csvoutput-json-csv-conv.csv") > 0) {\n\t\t#}\n\t\t#else {\n\t\tif "}

Is this the correct way to use 'ruby' filter plugin?

5

Thank you for your response. Just to make sure, the entire config file is executed for every line of the input file (json format), correct?

Yes.

So if my csv plugin outputs the output to a specific file path, the file will be created after the very first line of the input file is parsed, correct?

Logstash does not guarantee that event 1 has been flushed to all outputs before event 2 enters the filters. Doing so would result in extremely poor performance. In other words, it can take a while (perhaps hundreds of events) before your ruby filter "discovers" that the file has been created.

6

Thanks, I fixed the problem by using 'init' setting in my ruby filter.
Reference: https://stackoverflow.com/questions/36097121/logstash-csv-output-headers

7

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