Need to parse multi-lines log message

Elastic Stack Logstash
1

Hi everyone,

I am new on Grok. I am in stuck with multiple lines of log message below. Can anyone help me look at my grok. Thanks
The log message:
[timestamp: 1621431760] abort handler of pid 1823 thread 1848977280
*** Stacks of threads *** (current thread is 1848977280)
Stack of thread=1848977280, depth=3
main
shutdownServices
EMThriftServer::stop

logstash show

 "@version" => "1",
"@timestamp" => 2024-05-25T01:32:29.912Z,
     "event" => {
    "original" => "   Stack of thread=1848977280, depth=3"
},
       "ecs" => {
    "version" => "8.0.0"
},
      "tags" => [
    [0] "beats_input_codec_plain_applied",
    **[1] "_grokparsefailure"**

I tried: filter
{
grok {
match => { "message" => "[%{WORD}:%{SPACE}%{NUMBER:dts}]%{SPACE}(<?rest>%{GREEDYDATA}(?m))" }
}
date {
match => [ "dts", "ddHHmmssSSS" ]
target => "@timestamp"
timezone => "UTC"
}

2

logstash is consuming the multi-line log message one line at a time. If you are using filebeat you should use the multiline processing there to combine the lines into a single event.

3

Hi, this is my filebeat.yml. Can you advice if it is not correct.Thanks

filebeat.inputs:

filestream is an input for collecting log messages from files.

  • type: log

    Unique ID among all inputs, an ID is required.

    id: 1

    Change to true to enable this input configuration.

    enabled: true

    Paths that should be crawled and fetched. Glob based paths.

    paths:

    • C:\Filebeat\logs*.txt
      #- c:\programdata\elasticsearch\logs*

    Multiline log

    multiline.type: pattern
    multiline.pattern: '^['
    multiline.negate: false
    multiline.match: after

4

You can use something like this:

filebeat.yml

- type: filestream
  id: idlog
  enabled: true
  paths:
    - C:\Filebeat\logs*.txt
  parsers:
  - multiline:
      type: pattern
      pattern: '^\[timestamp:'
      negate: true
      match: after

logstash.conf contains extracted pid, thread and 2 part of message. I don't know a logic, which fields are important. Easiest is to put everything in the message after the timestamp. Of course you can parse depth, I assume services as fields.

input {
  beats {
    port => 5044
  }  
}

filter {

	grok {match => {
	message => "\[timestamp: %{POSINT:timestamp}\]%{SPACE}%{DATA:msgpart1}%{SPACE}pid %{POSINT:pid} thread %{POSINT:threadid}%{GREEDYDATA:msgpart2}"
		}    
	}
	date {
		match => [ "timestamp", "UNIX" ]
	}

}

output {
    stdout { codec => rubydebug{} }
}

Result:

{
      "threadid" => "1848977281",
     "timestamp" => "1621431765",
       "message" => "[timestamp: 1621431765] abort handler of pid 1111 thread 1848977281\n*** Stacks of threads *** (current thread is 1848977281)\nStack of thread=1848977281, depth=1\nmain\nshutdownServices\nEMThriftServer::stop",
    "@timestamp" => 2021-05-19T13:42:45.000Z,
      "@version" => "1",
           "pid" => "1111",
      "msgpart2" => "\n*** Stacks of threads *** (current thread is 1848977281)\nStack of thread=1848977281, depth=1\nmain\nshutdownServices\nEMThriftServer::stop",
      "msgpart1" => "abort handler of"
}
5

thanks @Rios and @Badger