IMPORTANT: No additional bug fixes or documentation updates will be released for this version. For the latest information, see the current release documentation.

« Filter plugins alter »

› ›

aggregate

edit

aggregate

edit

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 #1

edit

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 #2

edit

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 works

edit

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

Synopsis

edit

This plugin supports the following configuration options:

Required configuration options:

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

Available configuration options:

Setting	Input type	Required	Default value
`add_field`	hash	No	`{}`
`add_tag`	array	No	`[]`
`code`	string	Yes
`end_of_task`	boolean	No	`false`
`map_action`	string	No	`"create_or_update"`
`periodic_flush`	boolean	No	`false`
`remove_field`	array	No	`[]`
`remove_tag`	array	No	`[]`
`task_id`	string	Yes
`timeout`	number	No	`0`

Details

edit

`add_field`

edit

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}" }
      }
    }
[source,ruby]
    # 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_tag`

edit

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}" ]
      }
    }
[source,ruby]
    # 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).

`code`

edit

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_task`

edit

Value type is boolean
Default value is false

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

`map_action`

edit

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_flush`

edit

Value type is boolean
Default value is false

Call the filter flush method at regular interval. Optional.

`remove_field`

edit

Value type is array
Default value is []

If this filter is successful, remove arbitrary fields from this event. Fields names can be dynamic and include parts of the event using the %{field} Example:

    filter {
      aggregate {
        remove_field => [ "foo_%{somefield}" ]
      }
    }
[source,ruby]
    # 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_tag`

edit

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}" ]
      }
    }
[source,ruby]
    # 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_id`

edit

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.

Example value : "%{application}%{my_task_id}"

`timeout`

edit

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.

« Filter plugins alter »