ADD new filed to log from previous log

Elastic Stack Logstash
1

Hello Community,

I have a log file in CEF format and log start and end times as meta data in separate lines,

startTime:1541145880148
endTime:1541146222463
CEF:0|Incapsula|SIEMintegration|1|1|Normal|0| sourceServiceName=site123.abcd.info siteid=1509732 suid=50005477 requestClientApplication=Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.0 deviceFacility=mia src=12.12.12.12 caIP=13.13.13.13 ccode=IL tag=www.elvis.com cicode=Rehovot cs7=31.8969 cs7Label=latitude cs8=34.8186 cs8Label=longitude Customer=CEFcustomer123 ver=TLSv1.2 ECDHE-RSA-AES128-GCM-SHA256 start=1453290121336 request=site123.abcd.info/main.css ref=www.incapsula.com/lama requestmethod=GET cn1=200 app=HTTP deviceExternalID=33411452762204224 in=54 xff=44.44.44.44 cpt=443

in Logstash input it considering as 3 new messages, I have 1000's of logs in one log file with single startTime and endTime.

Now I want to append that start time and end time to each log of the logfile like below

startTime:1541145880148 endTime:1541146222463 CEF:0|Incapsula|SIEMintegration|1|1|Normal|0| sourceServiceName=site123.abcd.info siteid=1509732 suid=50005477 requestClientApplication=Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.0 deviceFacility=mia src=12.12.12.12 caIP=13.13.13.13 ccode=IL tag=www.elvis.com cicode=Rehovot cs7=31.8969 cs7Label=latitude cs8=34.8186 cs8Label=longitude Customer=CEFcustomer123 ver=TLSv1.2 ECDHE-RSA-AES128-GCM-SHA256 start=1453290121336 request=site123.abcd.info/main.css ref=www.incapsula.com/lama requestmethod=GET cn1=200 app=HTTP deviceExternalID=33411452762204224 in=54 xff=44.44.44.44 cpt=443

Here I am trying like use add_filed by catching when message contains startTime and create a new filed but I am not sure how I can append to next log because I believe when I close if condition it won't remember that filed to append to next log.

2

Unfortunately what you have is not a legal CEF-format log, but a log file that contains a mix of data in different formats, some lines of which are CEF, and others are a custom key/value format. This is going to make things difficult since no one codec is capable of processing all lines in a given log file.

This is made perhaps more difficult because Logstash (and likely your application that is doing the logging) does not make ordering guarantees, so there is no guarantee that a filter processing your events will see the start- and end-metadata events before receiving the related CEF log. It is also probable that your application will interleave messages from separate threads, which means you might get something like the following, where one start/end pair cannot be conclusively matched up with a specific CEF-line:

startTime:1541145880148
startTime:1541145880172
endTime:1541146222459
CEF:0|Incapsula|SIEMintegration|1|1|Normal|0| sourceServiceName=site123.abcd.info siteid=1509732 suid=50005477 requestClientApplication=Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.0 deviceFacility=mia src=12.12.12.12 caIP=13.13.13.13 ccode=IL tag=www.elvis.com cicode=Rehovot cs7=31.8969 cs7Label=latitude cs8=34.8186 cs8Label=longitude Customer=CEFcustomer123 ver=TLSv1.2 ECDHE-RSA-AES128-GCM-SHA256 start=1453290121336 request=site123.abcd.info/main.css ref=www.incapsula.com/lama requestmethod=GET cn1=200 app=HTTP deviceExternalID=33411452762204224 in=54 xff=44.44.44.44 cpt=443
endTime:1541146222463
CEF:0|Incapsula|SIEMintegration|1|1|Normal|0| sourceServiceName=site123.abcd.info siteid=1509732 suid=50005477 requestClientApplication=Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.0 deviceFacility=mia src=12.12.12.12 caIP=13.13.13.13 ccode=IL tag=www.elvis.com cicode=Rehovot cs7=31.8969 cs7Label=latitude cs8=34.8186 cs8Label=longitude Customer=CEFcustomer123 ver=TLSv1.2 ECDHE-RSA-AES128-GCM-SHA256 start=1453290121336 request=site123.abcd.info/main.css ref=www.incapsula.com/lama requestmethod=GET cn1=200 app=HTTP deviceExternalID=33411452762204224 in=54 xff=44.44.44.44 cpt=443
i

If possible, I would recommend the following:

  1. Emit the start and end timestamps within your CEF line, OR
  2. Emit the start and end timestamps in CEF format AND include a unique "request identifier" or "activity identifier" in a CEF extension field so that the logs belonging to the same request or activity can be grouped together after the fact.
3

@yaauie That logs is originally from Incapsula website CEF log format example and I intentionally not added all metadata except start and End times, Also, I guarantee on logstash processing from top to bottom and for each logfile I have only one start and end times at the top.

All I need is if I process the logline of startTime , I want to hold the variable parsed from that line and injest and remaining CEF formated logs in my logfile

4

Logstash itself does not guarantee that a given filter will receive events that have been produced by your inputs in order; you can increase the liklihood that events are processed in order by reducing the number of pipeline workers to 1 and setting the batch size to 1 (which will significantly degrade performance), but even then we still are not guaranteed to process the lines in order 100% of the time.

5

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