aggregateedit

Note

This is a community-maintained plugin! It does not ship with Logstash by default, but it is easy to install by running bin/logstash-plugin install logstash-filter-aggregate.

The aim of this filter is to aggregate information available among several events (typically log lines) belonging to a same task, and finally push aggregated information into final task event.

You should be very careful to set logstash filter workers to 1 (-w 1 flag) for this filter to work correctly otherwise documents may be processed out of sequence and unexpected results will occur.

Example #1edit

  • with these given logs :
 INFO - 12345 - TASK_START - start
 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end
  • you can aggregate "sql duration" for the whole task with this configuration :
 filter {
   grok {
     match => [ "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} - %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" ]
   }

   if [logger] == "TASK_START" {
     aggregate {
       task_id => "%{taskid}"
       code => "map['sql_duration'] = 0"
       map_action => "create"
     }
   }

   if [logger] == "SQL" {
     aggregate {
       task_id => "%{taskid}"
       code => "map['sql_duration'] += event['duration']"
       map_action => "update"
     }
   }

   if [logger] == "TASK_END" {
     aggregate {
       task_id => "%{taskid}"
       code => "event['sql_duration'] = map['sql_duration']"
       map_action => "update"
       end_of_task => true
       timeout => 120
     }
   }
 }
  • the final event then looks like :
{
       "message" => "INFO - 12345 - TASK_END - end message",
  "sql_duration" => 46
}

the field sql_duration is added and contains the sum of all sql queries durations.

Example #2edit

  • If you have the same logs than example #1, but without a start log :
 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end
  • you can also aggregate "sql duration" with a slightly different configuration :
 filter {
   grok {
     match => [ "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} - %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" ]
   }

   if [logger] == "SQL" {
     aggregate {
       task_id => "%{taskid}"
       code => "map['sql_duration'] ||= 0 ; map['sql_duration'] += event['duration']"
     }
   }

   if [logger] == "TASK_END" {
     aggregate {
       task_id => "%{taskid}"
       code => "event['sql_duration'] = map['sql_duration']"
       end_of_task => true
       timeout => 120
     }
   }
 }
  • the final event is exactly the same than example #1
  • the key point is the "||=" ruby operator. It allows to initialize sql_duration map entry to 0 only if this map entry is not already initialized

How it worksedit

  • the filter needs a "task_id" to correlate events (log lines) of a same task
  • at the task beggining, filter creates a map, attached to task_id
  • for each event, you can execute code using event and map (for instance, copy an event field to map)
  • in the final event, you can execute a last code (for instance, add map data to final event)
  • after the final event, the map attached to task is deleted
  • in one filter configuration, it is recommanded to define a timeout option to protect the feature against unterminated tasks. It tells the filter to delete expired maps
  • if no timeout is defined, by default, all maps older than 1800 seconds are automatically deleted
  • finally, if code execution raises an exception, the error is logged and event is tagged _aggregateexception

 

Synopsisedit

This plugin supports the following configuration options:

Required configuration options:

aggregate {
    code => ...
    task_id => ...
}

Available configuration options:

Detailsedit

 

add_fieldedit

  • Value type is hash
  • Default value is {}

If this filter is successful, add any arbitrary fields to this event. Field names can be dynamic and include parts of the event using the %{field}.

Example:

filter {
  aggregate {
    add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
  }
}
# You can also add multiple fields at once:
filter {
  aggregate {
    add_field => {
      "foo_%{somefield}" => "Hello world, from %{host}"
      "new_field" => "new_static_value"
    }
  }
}

If the event has field "somefield" == "hello" this filter, on success, would add field foo_hello if it is present, with the value above and the %{host} piece replaced with that value from the event. The second example would also add a hardcoded field.

add_tagedit

  • Value type is array
  • Default value is []

If this filter is successful, add arbitrary tags to the event. Tags can be dynamic and include parts of the event using the %{field} syntax.

Example:

filter {
  aggregate {
    add_tag => [ "foo_%{somefield}" ]
  }
}
# You can also add multiple tags at once:
filter {
  aggregate {
    add_tag => [ "foo_%{somefield}", "taggedy_tag"]
  }
}

If the event has field "somefield" == "hello" this filter, on success, would add a tag foo_hello (and the second example would of course add a taggedy_tag tag).

codeedit

  • This is a required setting.
  • Value type is string
  • There is no default value for this setting.

The code to execute to update map, using current event.

Or on the contrary, the code to execute to update event, using current map.

You will have a map variable and an event variable available (that is the event itself).

Example value : "map[sql_duration] += event[duration]"

end_of_taskedit

  • Value type is boolean
  • Default value is false

Tell the filter that task is ended, and therefore, to delete map after code execution.

map_actionedit

  • Value type is string
  • Default value is "create_or_update"

Tell the filter what to do with aggregate map.

create: create the map, and execute the code only if map wasn’t created before

update: doesn’t create the map, and execute the code only if map was created before

create_or_update: create the map if it wasn’t created before, execute the code in all cases

periodic_flushedit

  • Value type is boolean
  • Default value is false

Call the filter flush method at regular interval. Optional.

remove_fieldedit

  • Value type is array
  • Default value is []

If this filter is successful, remove arbitrary fields from this event. Example:

filter {
  aggregate {
    remove_field => [ "foo_%{somefield}" ]
  }
}
# You can also remove multiple fields at once:
filter {
  aggregate {
    remove_field => [ "foo_%{somefield}", "my_extraneous_field" ]
  }
}

If the event has field "somefield" == "hello" this filter, on success, would remove the field with name foo_hello if it is present. The second example would remove an additional, non-dynamic field.

remove_tagedit

  • Value type is array
  • Default value is []

If this filter is successful, remove arbitrary tags from the event. Tags can be dynamic and include parts of the event using the %{field} syntax.

Example:

filter {
  aggregate {
    remove_tag => [ "foo_%{somefield}" ]
  }
}
# You can also remove multiple tags at once:
filter {
  aggregate {
    remove_tag => [ "foo_%{somefield}", "sad_unwanted_tag"]
  }
}

If the event has field "somefield" == "hello" this filter, on success, would remove the tag foo_hello if it is present. The second example would remove a sad, unwanted tag as well.

task_idedit

  • This is a required setting.
  • Value type is string
  • There is no default value for this setting.

The expression defining task ID to correlate logs.

This value must uniquely identify the task in the system.

timeoutedit

  • Value type is number
  • Default value is 0

The amount of seconds after a task "end event" can be considered lost.

The task "map" is evicted.

Default value (0) means no timeout so no auto eviction.