Reading over multiple lines with grok

Elastic Stack Logstash
1

Hi
I have this Log where I want to match the start of a thread of device with the corresponding spoolid.

30 Jul 2019 09:56:28 <ID1> DEVICE0 START_THREAD THREAD_ID(B01234)
30 Jul 2019 09:56:28 <ID1> DEVICE1 START_THREAD THREAD_ID(B12345)
30 Jul 2019 09:56:28 <ID1> DEVICE2 START_THREAD THREAD_ID(A12345)
30 Jul 2019 09:56:28 <ID2> DEVICE1 PROCESSING SPOOLID 100 
30 Jul 2019 09:56:28 <ID2> DEVICE2 PROCESSING SPOOLID 101
30 Jul 2019 09:56:28 <ID2> DEVICE2 PROCESSING SPOOLID 101
30 Jul 2019 09:56:28 <ID2> DEVICE1 PROCESSING SPOOLID 100
30 Jul 2019 09:56:28 <ID4> DEVICE1 SPOOLID 100 PROCESSED
30 Jul 2019 09:56:28 <ID4> DEVICE2 SPOOLID 101 PROCESSED
30 Jul 2019 09:56:28 <ID3> DEVICE2 STOP_THREAD THREAD_ID(B12345) 
30 Jul 2019 09:56:28 <ID3> DEVICE2 STOP_THREAD THREAD_ID(A12345)
30 Jul 2019 09:56:28 <ID1> DEVICE2 START_THREAD THREAD_ID(A23456)
30 Jul 2019 09:56:29 <ID2> DEVICE2 PROCESSING SPOOLID 102
30 Jul 2019 09:56:29 <ID2> DEVICE2 PROCESSING SPOOLID 102
30 Jul 2019 09:56:29 <ID4> DEVICE2 SPOOLID 102 PROCESSED
30 Jul 2019 09:56:29 <ID3> DEVICE2 STOP_THREAD THREAD_ID(A23456) 
30 Jul 2019 09:56:29 <ID2> DEVICE0 PROCESSING SPOOLID 99 
30 Jul 2019 09:56:29 <ID4> DEVICE0 SPOOLID 99 PROCESSED
30 Jul 2019 09:56:29 <ID3> DEVICE0 STOP_THREAD THREAD_ID(B12345)

To get the corresponding fields I have this code:
(?m)%{DATA}\>\s+(?<device>\b%{DATA}\b)\s*START_THREAD%{DATA}\k<device>\s+SPOOLID\s+%{NUMBER:num}\s+PROCESSED

In the Heroku Grok Debugger I get the correct fields but not in my Logstash on my server because there I get line by line and therefore a _grokparsefailure error.

2

I would use an aggregate filter for that...

    dissect { mapping => { "message" => "%{[@metadata][ts]} %{+[@metadata][ts]} %{+[@metadata][ts]} %{+[@metadata][ts]} <%{id}> %{dev} %{restOfLine}" } }
    date { match => [ "[@metadata][ts]", "dd MMM YYYY HH:mm:ss" ] }
    grok {
        match => {
            "restOfLine" => [
                "^START_THREAD THREAD_ID\(%{DATA:thread}\)",
                "^STOP_THREAD THREAD_ID\(%{DATA:thread}\)",
                "^PROCESSING SPOOLID %{NUMBER:spool}",
                "^SPOOLID %{NUMBER:spool} PROCESSED"
            ]
        }
    }
    if " START_THREAD " in [message] {
        aggregate {
            task_id => "%{dev}"
            code => 'map["thread"] = event.get("thread")'
            map_action => "create"
        }
    }
    if " PROCESSED" in [message] {
        aggregate {
            task_id => "%{dev}"
            code => "event.set('thread', map['thread'])"
            map_action => "update"
            end_of_task => true
            timeout => 120
        }
    } else {
        drop {}
    }
3

Thank you for your reply. That is also a way, but it is still not clear for me why the (?m) in my grok code is not recognized. Is there a special configuration needed in logstash.yml for this to work?

4

I don't think you can do it in grok. If you use a multiline codec on the input to ingest the complete set of lines as one event then the grok pattern matches, but only once

   "message" => "30 Jul 2019 09:56:28 <ID1> DEVICE0 START_THREAD THREAD_ID(B01234)\n30 Jul 2019 09:56:28 <ID1> DEVICE1 START_THREAD THREAD_ID(B12345)\n30 Jul 2019 09:56:28 <ID1> DEVICE2 START_THREAD THREAD_ID(A12345)\n30 Jul 2019 09:56:28 <ID2> DEVICE1 PROCESSING SPOOLID 100 \n30 Jul 2019 09:56:28 <ID2> DEVICE2 PROCESSING SPOOLID 101\n30 Jul 2019 09:56:28 <ID2> DEVICE2 PROCESSING SPOOLID 101\n30 Jul 2019 09:56:28 <ID2> DEVICE1 PROCESSING SPOOLID 100\n30 Jul 2019 09:56:28 <ID4> DEVICE1 SPOOLID 100 PROCESSED\n30 Jul 2019 09:56:28 <ID4> DEVICE2 SPOOLID 101 PROCESSED\n30 Jul 2019 09:56:28 <ID3> DEVICE2 STOP_THREAD THREAD_ID(B12345) \n30 Jul 2019 09:56:28 <ID3> DEVICE2 STOP_THREAD THREAD_ID(A12345)\n30 Jul 2019 09:56:28 <ID1> DEVICE2 START_THREAD THREAD_ID(A23456)\n30 Jul 2019 09:56:29 <ID2> DEVICE2 PROCESSING SPOOLID 102\n30 Jul 2019 09:56:29 <ID2> DEVICE2 PROCESSING SPOOLID 102\n30 Jul 2019 09:56:29 <ID4> DEVICE2 SPOOLID 102 PROCESSED\n30 Jul 2019 09:56:29 <ID3> DEVICE2 STOP_THREAD THREAD_ID(A23456) \n30 Jul 2019 09:56:29 <ID2> DEVICE0 PROCESSING SPOOLID 99 \n30 Jul 2019 09:56:29 <ID4> DEVICE0 SPOOLID 99 PROCESSED\n30 Jul 2019 09:56:29 <ID3> DEVICE0 STOP_THREAD THREAD_ID(B12345)",
    "device" => "DEVICE0",
       "num" => "99"
5

I need the device and the spoolid for the task_id, because the different spoolid for the same device can be mixed even before one device has finished processing . Howerver when I do it like this I do not get the thread number which I also need.
Is there a solution?