elasticsearch/logstash-contrib — pull request #55 by fbaligand (2014-05-09 15:17:44+00:00)
Title: new plugin : aggregate filter

**New plugin : aggregate filter**

*This pull request is linked to this jira issue :*
https://logstash.jira.com/browse/LOGSTASH-2194

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 :
```
INFO - 12345 - TASK_START - start message
INFO - 12345 - DAO - MyDao.findById - 12
INFO - 12345 - DAO - MyDao.findAll - 34
INFO - 12345 - TASK_END - end message
```
* you can aggregate "dao duration" with this configuration :
``` ruby
filter {
grok {
match => [ "message", "%{SPACE}%{LOGLEVEL:loglevel} - %{NOTSPACE:requestid} - %{NOTSPACE:logger} - %{GREEDYDATA:msg}(\n%{GREEDYDATA})?" ]
}

if [logger] == "TASK_START" {
aggregate {
task_id => "%{requestid}"
code => "map['dao.duration'] = 0"
map_action => "create"
}
}

if [logger] == "DAO" {
grok {
match => [ "msg", "%{JAVACLASS:dao_call} - %{INT:duration:int}" ]
}
aggregate {
task_id => "%{requestid}"
code => "map['dao.duration'] += event['duration']"
map_action => "update"
}
}

if [logger] == "TASK_END" {
aggregate {
task_id => "%{requestid}"
code => "event.to_hash.merge!(map)"
map_action => "update"
end_of_task => true
timeout => 120
}
}
}
```