› ›

聚合过滤器插件

编辑

聚合过滤器插件

编辑

插件版本：v2.10.0
发布日期：2021-10-11
变更日志

其他版本，请参见版本化插件文档。

获取帮助

编辑

如有任何关于插件的问题，请在 Discuss 论坛中发帖。如发现错误或有功能需求，请在 Github 中提交问题。如需查看 Elastic 支持的插件列表，请参阅 Elastic 支持矩阵。

描述

编辑

此过滤器的目的是聚合属于同一任务的多个事件（通常是日志行）中的可用信息，最终将聚合信息推送到最终任务事件中。

为了使此过滤器正常工作，您应非常小心地将 Logstash 过滤器工作线程设置为 1（-w 1 标志），否则事件可能会乱序处理，并导致意外结果。

示例 #1

编辑

给定以下日志

 INFO - 12345 - TASK_START - start
 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end

您可以使用以下配置聚合整个任务的“sql duration”

 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.get('duration')"
       map_action => "update"
     }
   }

   if [logger] == "TASK_END" {
     aggregate {
       task_id => "%{taskid}"
       code => "event.set('sql_duration', map['sql_duration'])"
       map_action => "update"
       end_of_task => true
       timeout => 120
     }
   }
 }

最终事件如下所示

{
  "message" => "INFO - 12345 - TASK_END - end message",
  "sql_duration" => 46
}

sql_duration 字段已添加，其中包含所有 sql 查询持续时间的总和。

示例 #2：没有起始事件

编辑

如果您有与示例 #1 相同的日志，但没有起始日志

 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end

您也可以使用略微不同的配置来聚合“sql duration”

 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.get('duration')"
     }
   }

   if [logger] == "TASK_END" {
     aggregate {
       task_id => "%{taskid}"
       code => "event.set('sql_duration', map['sql_duration'])"
       end_of_task => true
       timeout => 120
     }
   }
 }

最终事件与示例 #1 完全相同
关键在于 "||=" ruby 运算符。它只允许在该 map 条目尚未初始化的情况下，将 *sql_duration* map 条目初始化为 0。

示例 #3：没有结束事件

编辑

第三个用例：您没有特定的结束事件。

一个典型的案例是聚合或跟踪用户行为。我们可以通过事件跟踪用户的 ID，但是一旦用户停止交互，事件就会停止传入。没有指示用户交互结束的特定事件。

在这种情况下，我们可以启用 *push_map_as_event_on_timeout* 选项，以便在超时发生时将聚合映射作为新事件推送。此外，我们可以启用 *timeout_code* 来在填充的超时事件上执行代码。我们还可以添加 *timeout_task_id_field*，以便我们可以关联 task_id，在本例中为用户 ID。

给定这些日志

INFO - 12345 - Clicked One
INFO - 12345 - Clicked Two
INFO - 12345 - Clicked Three

您可以像这样聚合用户点击次数

filter {
  grok {
    match => [ "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:user_id} - %{GREEDYDATA:msg_text}" ]
  }

  aggregate {
    task_id => "%{user_id}"
    code => "map['clicks'] ||= 0; map['clicks'] += 1;"
    push_map_as_event_on_timeout => true
    timeout_task_id_field => "user_id"
    timeout => 600 # 10 minutes timeout
    timeout_tags => ['_aggregatetimeout']
    timeout_code => "event.set('several_clicks', event.get('clicks') > 1)"
  }
}

十分钟后，这将产生一个类似这样的事件

{
  "user_id": "12345",
  "clicks": 3,
  "several_clicks": true,
    "tags": [
       "_aggregatetimeout"
    ]
}

示例 #4：没有结束事件，任务一个接一个地到来

编辑

第四个用例：与示例 #3 相同，没有特定的结束事件，而且任务一个接一个地到来。

也就是说：任务不会交错。所有 task1 事件都来了，然后所有 task2 事件都来了……

在这种情况下，您不想等待任务超时来刷新聚合映射。

一个典型的案例是聚合来自 jdbc 输入插件的结果。
假设您有以下 SQL 查询：SELECT country_name, town_name FROM town
使用 jdbc 输入插件，您可以从以下内容获取这 3 个事件

  { "country_name": "France", "town_name": "Paris" }
  { "country_name": "France", "town_name": "Marseille" }
  { "country_name": "USA", "town_name": "New-York" }

并且您希望将这两个结果事件推送到 Elasticsearch

  { "country_name": "France", "towns": [ {"town_name": "Paris"}, {"town_name": "Marseille"} ] }
  { "country_name": "USA", "towns": [ {"town_name": "New-York"} ] }

您可以使用 push_previous_map_as_event 聚合插件选项来实现此目的

   filter {
     aggregate {
       task_id => "%{country_name}"
       code => "
         map['country_name'] ||= event.get('country_name')
         map['towns'] ||= []
         map['towns'] << {'town_name' => event.get('town_name')}
         event.cancel()
       "
       push_previous_map_as_event => true
       timeout => 3
     }
   }

关键在于，每次聚合插件检测到新的 country_name 时，它都会将之前的聚合映射作为新的 Logstash 事件推送，然后为下一个国家/地区创建一个新的空映射
当 3 秒超时到来时，最后一个聚合映射将作为新事件推送
初始事件（未聚合的事件）由于无用而被丢弃（感谢 event.cancel()）
最后一点：如果并非每个事件都填充了一个字段（例如“town_postcode”字段），则 ||= 运算符将允许您将第一个“非空”值推入聚合映射。例如：map['town_postcode'] ||= event.get('town_postcode')

示例 #5：没有结束事件，并尽快推送事件

编辑

第五个用例：与示例 #3 相同，没有结束事件。

事件会无限期地持续到来，您希望在最后一次用户交互后尽快推送聚合映射，而无需等待 timeout。

这允许将聚合事件更接近实时地推送。

一个典型的案例是聚合或跟踪用户行为。

我们可以通过事件跟踪用户的 ID，但是一旦用户停止交互，事件就会停止传入。

没有指示用户交互结束的特定事件。

当在指定的用户（task_id）没有事件到达超过指定的 inactivity_timeout 秒时，将认为用户交互已结束。

如果用户持续交互的时间超过 timeout 秒（自第一个事件以来），则聚合映射仍将在超时发生时被删除并作为新事件推送。

与示例 #3 的区别在于，事件将在用户停止交互 inactivity_timeout 秒后立即推送，而不是等待自第一个事件以来的 timeout 秒结束。

在这种情况下，我们可以启用 *push_map_as_event_on_timeout* 选项，以便在非活动超时发生时将聚合映射作为新事件推送。

此外，我们可以启用 *timeout_code* 来在填充的超时事件上执行代码。

我们还可以添加 *timeout_task_id_field*，以便我们可以关联 task_id，在本例中为用户 ID。

给定这些日志

INFO - 12345 - Clicked One
INFO - 12345 - Clicked Two
INFO - 12345 - Clicked Three

您可以像这样聚合用户点击次数

filter {
 grok {
   match => [ "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:user_id} - %{GREEDYDATA:msg_text}" ]
 }
 aggregate {
   task_id => "%{user_id}"
   code => "map['clicks'] ||= 0; map['clicks'] += 1;"
   push_map_as_event_on_timeout => true
   timeout_task_id_field => "user_id"
   timeout => 3600 # 1 hour timeout, user activity will be considered finished one hour after the first event, even if events keep coming
   inactivity_timeout => 300 # 5 minutes timeout, user activity will be considered finished if no new events arrive 5 minutes after the last event
   timeout_tags => ['_aggregatetimeout']
   timeout_code => "event.set('several_clicks', event.get('clicks') > 1)"
 }
}

在五分钟的非活动时间或自第一个事件以来的一小时后，这将产生一个类似这样的事件

{
 "user_id": "12345",
 "clicks": 3,
 "several_clicks": true,
   "tags": [
      "_aggregatetimeout"
   ]
}

工作原理

编辑

过滤器需要一个“task_id”来关联同一任务的事件（日志行）
在任务开始时，过滤器创建一个与 task_id 关联的映射
对于每个事件，您可以使用 *event* 和 *map* 执行代码（例如，将事件字段复制到映射）
在最终事件中，您可以执行最后一段代码（例如，将映射数据添加到最终事件）
在最终事件之后，与任务关联的映射将被删除（感谢 end_of_task => true）
一个聚合映射与一个 task_id 值绑定，该值与一个 task_id 模式绑定。因此，如果您有 2 个具有不同 task_id 模式的过滤器，即使您具有相同的 task_id 值，它们也不会共享相同的聚合映射。
在一个过滤器配置中，建议定义一个超时选项来保护该功能免受未终止的任务的影响。它告诉过滤器删除已过期的映射
如果未定义超时，则默认情况下，所有超过 1800 秒的映射都会自动删除
所有超时选项都必须仅在一个针对每个 task_id 模式的聚合过滤器（每个管道）中定义。超时选项包括：timeout、inactivity_timeout、timeout_code、push_map_as_event_on_timeout、push_previous_map_as_event、timeout_timestamp_field、timeout_task_id_field、timeout_tags
如果 code 执行引发异常，则会记录错误并标记事件为 * _aggregateexception*

用例

编辑

从任务日志中提取一些不错的指标，并将它们推送到最终任务日志事件中（如示例 #1 和 #2 中所示）
提取任何任务日志行中的错误信息，并将其推送到最终任务事件中（如果存在任何错误信息，则获取包含所有错误信息的最终事件）
将所有后端调用作为列表提取，并将此列表推送到最终任务事件中（以获取任务配置文件）
提取几行中记录的所有 http 标头，并将此列表推送到最终任务事件中（完整的 http 请求信息）
对于每个后端调用，收集可在几行中获得的调用详细信息，对其进行分析，最后标记最终后端调用日志行（错误、超时、业务警告……）
最后，任务 ID 可以是满足您需求的任何相关 ID：它可以是会话 ID、文件路径……

聚合过滤器配置选项

编辑

此插件支持以下配置选项以及稍后描述的常用选项。

设置	输入类型	必需
`aggregate_maps_path`	字符串，一个有效的系统文件路径	否
`code`	字符串	是
`end_of_task`	布尔值	否
`inactivity_timeout`	数字	否
`map_action`	字符串，`["create", "update", "create_or_update"]` 之一	否
`push_map_as_event_on_timeout`	布尔值	否
`push_previous_map_as_event`	布尔值	否
`task_id`	字符串	是
`timeout`	数字	否
`timeout_code`	字符串	否
`timeout_tags`	数组	否
`timeout_task_id_field`	字符串	否
`timeout_timestamp_field`	字符串	否

另请参见常用选项，了解所有过滤器插件支持的选项列表。

`aggregate_maps_path`

编辑

值类型为字符串
此设置没有默认值。

Logstash 停止时存储聚合映射并从 Logstash 启动时加载聚合映射的文件路径。

如果未定义，则不会在 Logstash 停止时存储聚合映射，并将丢失。必须在每个管道中的一个聚合过滤器中定义（因为聚合映射在管道级别共享）。

示例

    filter {
      aggregate {
        aggregate_maps_path => "/path/to/.aggregate_maps"
      }
    }

`code`

编辑

这是一个必需的设置。
值类型为字符串
此设置没有默认值。

使用当前事件执行以更新聚合映射的代码。

或者相反，使用聚合映射执行以更新事件的代码。

可用变量为

event：当前 Logstash 事件

map：与 task_id 关联的聚合映射，包含键值对。数据结构是 ruby 哈希表

map_meta：与聚合映射关联的元信息。它允许设置自定义的timeout或inactivity_timeout。它也允许获取creation_timestamp、lastevent_timestamp和task_id。

new_event_block：用于发出新的Logstash事件的块。请参阅第二个示例了解如何使用它。

当选项push_map_as_event_on_timeout=true时，如果在code块中设置map_meta.timeout=0，则聚合映射将立即作为新事件推送。

示例

    filter {
      aggregate {
        code => "map['sql_duration'] += event.get('duration')"
      }
    }

要在代码执行期间创建其他事件并立即发出，可以使用new_event_block.call(event)函数，如下例所示。

    filter {
      aggregate {
        code => "
            data = {:my_sql_duration => map['sql_duration']}
            generated_event = LogStash::Event.new(data)
            generated_event.set('my_other_field', 34)
            new_event_block.call(generated_event)
        "
      }
    }

函数new_event_block.call的参数必须是LogStash::Event类型。要创建这样的对象，可以使用相同类的构造函数：LogStash::Event.new()。LogStash::Event.new()可以接收ruby Hash类型的参数来初始化新的事件字段。

`end_of_task`

编辑

值类型为布尔值
默认值为false

告诉过滤器任务已结束，因此在代码执行后删除聚合映射。

`inactivity_timeout`

编辑

值类型为数值
此设置没有默认值。

自上次事件以来被视为过期任务的秒数。

当任务超时时，其聚合映射将被清除。

如果push_map_as_event_on_timeout或push_previous_map_as_event设置为true，则任务聚合映射将作为新的Logstash事件推送。

可以为每个“task_id”模式定义inactivity_timeout。

inactivity_timeout必须小于timeout。

`map_action`

编辑

值类型为字符串
默认值为"create_or_update"

告诉过滤器如何处理聚合映射。

"create"：创建映射，并且仅在之前未创建映射时才执行代码。

"update"：不创建映射，并且仅在之前已创建映射时才执行代码。

"create_or_update"：如果之前未创建映射则创建映射，在所有情况下都执行代码。

`push_map_as_event_on_timeout`

编辑

值类型为布尔值
默认值为false

启用此选项后，每次检测到任务超时时，它都会将任务聚合映射作为新的Logstash事件推送。这使得能够在Logstash中检测和处理任务超时，还可以管理没有显式结束事件的任务。

`push_previous_map_as_event`

编辑

值类型为布尔值
默认值为false

启用此选项后，每次聚合插件检测到新的task id时，它都会将之前的聚合映射作为新的Logstash事件推送，然后为下一个任务创建一个新的空映射。

此选项仅在任务一个接一个地到来时才能正常工作。这意味着：所有task1事件，然后所有task2事件，等等…

`task_id`

编辑

这是一个必需的设置。
值类型为字符串
此设置没有默认值。

定义任务ID以关联日志的表达式。

此值必须唯一标识任务。

示例

    filter {
      aggregate {
        task_id => "%{type}%{my_task_id}"
      }
    }

`timeout`

编辑

值类型为数值
默认值为1800

自第一个事件以来被视为过期任务的秒数。

当任务超时时，其聚合映射将被清除。

如果push_map_as_event_on_timeout或push_previous_map_as_event设置为true，则任务聚合映射将作为新的Logstash事件推送。

可以为每个“task_id”模式定义超时。

`timeout_code`

编辑

值类型为字符串
此设置没有默认值。

当'push_map_as_event_on_timeout'或'push_previous_map_as_event'设置为true时，执行以完成超时生成的事件的代码。代码块将可以访问预先填充了聚合映射的新的超时生成的事件。

如果设置了'timeout_task_id_field'，则事件也会填充task_id值。

示例

    filter {
      aggregate {
        timeout_code => "event.set('state', 'timeout')"
      }
    }

`timeout_tags`

编辑

值类型为数组
默认值为[]

定义在生成并产生超时事件时要添加的标签。

示例

    filter {
      aggregate {
        timeout_tags => ["aggregate_timeout"]
      }
    }

`timeout_task_id_field`

编辑

值类型为字符串
此设置没有默认值。

此选项指示超时生成的事件的字段，其中将设置当前“task_id”值。这有助于关联哪些任务已超时。

默认情况下，如果未设置此选项，则不会将task id值设置到超时生成的事件中。

示例

    filter {
      aggregate {
        timeout_task_id_field => "task_id"
      }
    }

`timeout_timestamp_field`

编辑

值类型为字符串
此设置没有默认值。

默认情况下，超时是使用Logstash运行的系统时间计算的。

设置此选项后，超时将使用此选项中指示的事件时间戳字段计算。这意味着当第一个事件到达聚合过滤器并导致映射创建时，映射创建时间将等于此事件时间戳。然后，每次新的事件到达聚合过滤器时，都会将事件时间戳与映射创建时间进行比较，以检查是否发生了超时。

此选项在使用选项push_map_as_event_on_timeout => true处理旧日志时特别有用。它允许根据旧日志上的超时生成聚合事件，其中系统时间不适用。

警告：为了使此选项正常工作，必须在第一个聚合过滤器上设置它。

示例

    filter {
      aggregate {
        timeout_timestamp_field => "@timestamp"
      }
    }

通用选项

编辑

所有过滤器插件都支持这些配置选项。

设置	输入类型	必需
`add_field`	hash	否
`add_tag`	数组	否
`enable_metric`	布尔值	否
`id`	字符串	否
`periodic_flush`	布尔值	否
`remove_field`	数组	否
`remove_tag`	数组	否

`add_field`

编辑

值类型为哈希表
默认值为{}

如果此过滤器成功，则向此事件添加任意字段。字段名称可以是动态的，并包含使用%{field}的事件部分。

示例

    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"
        }
      }
    }

如果事件具有字段"somefield" == "hello"，则此过滤器成功后，如果存在，将添加字段foo_hello，其值为上述值，并且%{host}部分被事件中的该值替换。第二个示例还将添加一个硬编码字段。

`add_tag`

编辑

值类型为数组
默认值为[]

如果此过滤器成功，则向事件添加任意标签。标签可以是动态的，并包含使用%{field}语法的事件部分。

示例

    filter {
      aggregate {
        add_tag => [ "foo_%{somefield}" ]
      }
    }

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

如果事件具有字段"somefield" == "hello"，则此过滤器成功后，将添加标签foo_hello（第二个示例当然会添加一个taggedy_tag标签）。

`enable_metric`

编辑

值类型为布尔值
默认值为true

禁用或启用此特定插件实例的指标日志记录。默认情况下，我们记录所有可以记录的指标，但您可以禁用特定插件的指标收集。

`id`

编辑

值类型为字符串
此设置没有默认值。

向插件配置添加唯一的ID。如果未指定ID，Logstash将生成一个。强烈建议在配置中设置此ID。当您有两个或多个相同类型的插件时，这尤其有用，例如，如果您有两个聚合过滤器。在这种情况下，添加一个命名ID将有助于在使用监控API时监控Logstash。

    filter {
      aggregate {
        id => "ABC"
      }
    }

id字段中的变量替换仅支持环境变量，不支持使用密钥存储中的值。

`periodic_flush`

编辑

值类型为布尔值
默认值为false

定期调用过滤器刷新方法。可选。

`remove_field`

编辑

值类型为数组
默认值为[]

如果此过滤器成功，则从此事件中删除任意字段。字段名称可以是动态的，并包含使用%{field}的事件部分示例

    filter {
      aggregate {
        remove_field => [ "foo_%{somefield}" ]
      }
    }

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

如果事件具有字段"somefield" == "hello"，则此过滤器成功后，如果存在，将删除名称为foo_hello的字段。第二个示例将删除另一个非动态字段。

`remove_tag`

编辑

值类型为数组
默认值为[]

如果此过滤器成功，则从事件中删除任意标签。标签可以是动态的，并包含使用%{field}语法的事件部分。

示例

    filter {
      aggregate {
        remove_tag => [ "foo_%{somefield}" ]
      }
    }

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

如果事件具有字段"somefield" == "hello"，则此过滤器成功后，如果存在，将删除标签foo_hello。第二个示例也将删除一个令人不快的、不需要的标签。

« 年龄过滤器插件修改过滤器插件 »