Merging two lines with the same id to use as one document

Elastic Stack Logstash
1

Hi,

I'm struggling a few days in order to use aggregate filter to work (or i think it should work like this)

I have two lines, as follows

E.g:

08:00:45,585|73560708|ibwsin-tp-359|INFO|LoggerService||LogExecutionResult[81249983]|600BC03800C739DBCA030C000520|search-subscription|request| <?xml version="1.0" .*blablabla|0|

08:00:45,838|73560961|ibwsin-tp-359|INFO|end||Dispatcher[1944359640]|600BC03800C739DBCA030C000520|END|255.309853ms|OK

I'd like to merge these two lines, specifically get duration of this request (255.309853ms) and put it on a document in order to generate visualization into kibana.

I tried to do this configuration on logstash (apparently everything is ok), but i do not see how to agregate these data.

Here's my config:

filter {
 mutate {
        gsub => [ "message" , "[|]" , " "]
}
grok {
    match => { "message" => "%{TIME:logTime} %{INT:random} %{NOTSPACE:thread} %{WORD:info} %{WORD:loggerservice} %{SPACE} %{NOTSPACE:logExecutionResult} %{NOTSPACE:tid} %{NOTSPACE:Service} %{WORD:loggerType} (?<dataXML>.*\<\/ib\-msg\>) %{GREEDYDATA:line}"}
    match => { "message" => "%{TIME:logTime} %{INT:random} %{NOTSPACE:thread} %{WORD:info} %{WORD:loggerservice} %{SPACE} %{NOTSPACE:logExecutionResult} %{NOTSPACE:tid} %{WORD:loggerType} %{NOTSPACE:duration} %{GREEDYDATA:line}"}
}
mutate {
       gsub => ["duration","ms",""]
       convert => { "duration" => "float" }
}
   if [loggerType] == "request" {
 aggregate {
   task_id => "%{tid}"
   code => "map['duration'] = 0"
   map_action => "create"
 }
   } else if [loggerType] == "END" {
 aggregate {
   task_id => "%{tid}"
   map_action => "update"
   code => "map['duration'] = event.get('duration')"
   end_of_task => true
   timeout => 10
 }
}
2

You could either save all the fields of the first line in map, and then drop that event, and add the contents of map to the event when processing the second line, or you stop using aggregate and do a bulk update, as described here.

3

Hi,

So your suggestion, is that i use every useful field from the first line do be mapped on aggregate like this?

if [loggerType] == "request" {
aggregate {
    task_id => "%{tid}"
    code => "map['duration'] = 0"
    map_action => "create"
    code => "map['Service'] = event.get('Service')"
    map_action => "create"
    ...
}

And then, use something like this?

if "request" in [tags] {
        drop { }
}
4

Hi,

I was able to use what you suggested, i'm kinda neubie using map.
Thanks for your help

if [loggerType] == "request" {
  aggregate {
     task_id => "%{tid}"
     code =>
     "map['flow'] = event.get('flow')
      map['Service'] = event.get('Service')
      map['dataXML'] = event.get('dataXML')
      map['duration'] = 0"
  }
  drop {}
} else if [loggerType] == "response" {
 aggregate {
   task_id => "%{tid}"
   code => "map['responseXML'] = event.get('responseXML')"
 }
 drop {}
} else if [loggerType] == "END" {
 aggregate {
   task_id => "%{tid}"
   code => "
   event.set('flow', map['flow'])
   event.set('Service', map['Service'])
   event.set('dataXML', map['dataXML'])
   event.set('responseXML', map['responseXML'])"
   end_of_task => true
   push_previous_map_as_event => true
   timeout => 3600
   timeout_tags => ["aggregate"]
 }
}
5

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