Logstash aggregate/mutate

Elastic Stack Logstash
1

Hey all, I'm attempting to combine take 2 logs and add a field from a previous log to the other log.
I've read around about aggregate and mutate, and all that but I'm not quite understanding how to do this.

I have 2 logs like so:

{"event":"eventX","id":"123456","name":"abcd"}
{"event":"eventZ","id":"123456","field":"value","otherfield":"another_value"}

These two things have an id in common, along with other log lines but I only want to add the name field from eventX, to the eventZ log line so it would all come together as:

{"event":"eventZ","id":"123456","field":"value","otherfield":"another_value", "name": "abcd"}

I figure I could use memorize, or maybe aggregate/mutate and add_field to do this but I can't figure out how to do so correctly.

Any help is appreciated, thanks!

2

You can do that with an aggregate filter. If the two lines are always in that order then do something like example 1.

If the order can vary then something similar to example 3. Note that only things you put into the map are added to the event that is created after the timeout is triggered.

Make sure pipeline.workers set to 1 and you may need to disable java_execution.

3

Righto, so I've got this so far.

filter {
   json {
     source => "context"
     target => "json_data"
   }

   if [json_data.event] == "eventX" {
     aggregate {
       task_id => "%{json_data.id}"
         code => "
           if json_data['name'].is_a?(String)
             map['name'] = json_data['name']
           end
         "
         map_action => "create"
     }
   }

   if [json_data.event] == "eventZ" {
     aggregate {
       task_id => "%{json_data.id}"
         code => "
         if map['name'].is_a?(String)
           event.set('name', map['name'])
         end
         "
         map_action => "create"
         end_of_task => true
         timeout => 10
     }
   }
 }

For reference, context is an inner json object, contains all the data from the above post.
So,
{"field": "value", "context": { "name": "abcd", ...
In this case, I get class org.jruby.RubyHash cannot be cast to class org.jruby.RubyIO which I know what it means, but don't know why it's happening.

(EDIT: I've also tried json.name as well, to no avail in case that might've been an issue.

Sorry for the ignorance.

4

Your syntax to access your field is wrong: https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#logstash-config-field-references
And just like you use event.set(…) to create/update field values with ruby, you need to use event.get(…) to read them.
And your map_action cannot be "create" if you want to work with an already existing map.
(And if you don't want to keep both the enriched eventZ and eventX, you'll need to cancel eventX)

5

What produces that exception?

6

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