I have developed a new filter plugin, which name is : aggregate.
I created the following pull-request to add it on logstash-contrib project :
Plugin Details :
The aim of this filter is to aggregate informations available among several events (typically log lines) belonging to a same task,
and finally push aggregated information into final task event.
To do that :
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
An example of use can be:
with this given data :
you can aggregate "dao duration" with this configuration :