使用处理器过滤和增强数据.md

您使用case，可能只需要从filebeat中导出的数据的子集，或者只需要增强导出的数据（例如，添加metadata）。

您可以配置每个 intput 包含或排除特定行或文件。允许您为每个input指定不同的过滤条件。

您可以使用filebeat.inputs部分下的include_lines、exclude_lines和exclude_files选项（请参阅 Inputs）。

这种方法的缺点是您需要为所需的每个过滤条件实现一个配置选项。

另一种方法（此处描述的方法）是定义处理器，对FileBeat导出的所有数据配置全局处理。

处理器

可定义处理器，在事件发送到output之前进行处理，libbeat库为以下方面提供了处理器：

• 减少导出字段的数量
• 使用额外的元数据增强事件
• 执行额外的处理和解码

每个处理器（processor）接受一个事件（event），对该事件应用定义的操作（action），然后返回事件。如果定义了多个事件，它们按照定义的顺序顺序执行。

event -> processor 1 -> event1 -> processor 2 -> event2 ...

删除事件 示例

删除所有DEBUG消息：

processors:
  - drop_event: 
      when:
        regexp:
          message: "^DBG:"

要删除来自某个日志文件的所有日志消息：

processors:
  - drop_event:
      when:
        contains:
          source: "test"

解码JSON 示例

filebeat导出的字段中有一个inner字段，其值是json字符串：

{ "outer": "value", "inner": "{\"data\": \"value\"}" }

以下配置对 inner JSON字符串进行解码：

filebeat.inputs:
- type: log
  paths:
    - input.json
  json.keys_under_root: true

processors:
  - decode_json_fields:
      fields: ["inner"]

output.console.pretty: true

结果输出如下所示：

{
  "@timestamp": "2016-12-06T17:38:11.541Z",
  "beat": {
    "hostname": "host.example.com",
    "name": "host.example.com",
    "version": "8.5.1"
  },
  "inner": {
    "data": "value"
  },
  "input": {
    "type": "log",
  },
  "offset": 55,
  "outer": "value",
  "source": "input.json",
  "type": "log"
}

定义处理器

要定义处理器，需要指定处理器名称、可选条件和一组参数：

processors:
  - <processor_name>:
      when:
        <condition>
      <parameters>

  - <processor_name>:
      when:
        <condition>
      <parameters>

...

• <processor_name>：指定处理器。
• <condition>：指定条件，如果条件存在，则仅在条件满足时执行操作。如果没有设置条件，则始终执行操作。
• <parameters>：要传递给处理器的参数列表。

更复杂的条件处理可以通过使用 if-then-else 处理器配置来完成。这允许基于单个条件执行多个处理器。

processors:
  - if:
      <condition>
    then: 
      - <processor_name>:
          <parameters>
      - <processor_name>:
          <parameters>
      ...
    else: 
      - <processor_name>:
          <parameters>
      - <processor_name>:
          <parameters>
      ...

处理器作用位置

• 配置的顶层，应用于filebeat收集的所有数据

• 配置在指定的input下，应用于该input收集的数据

  - type: <input_type>
    processors:
      - <processor_name>:
          when:
            <condition>
          <parameters>
  ...

  相似的，对于filebeat模块，您可以在模块定义的input部分下定义处理器。

处理器

支持的处理器：

• add_cloud_metadata
• add_cloudfoundry_metadata
• add_docker_metadata
• add_fields
• add_host_metadata
• add_id
• add_kubernetes_metadata
• add_labels
• add_locale
• add_nomad_metadata
• add_observer_metadata
• add_process_metadata
• add_tags
• community_id
• convert
• copy_fields
• decode_base64_field
• decode_cef
• decode_csv_fields
• decode_json_fields
• decode_xml
• decode_xml_wineventlog
• decompress_gzip_field
• detect_mime_type
• dissect
• dns
• drop_event
• drop_fields
• extract_array
• fingerprint
• include_fields
• rate_limit
• registered_domain
• rename
• script
• syslog
• timestamp
• translate_sid
• truncate_fields
• urldecode

条件

每个条件接收一个要比较的字段。您可以通过在字段之间使用AND来指定相同条件下的多个字段（例如，field1 AND field2）。

对于每个字段，您可以指定一个简单的字段名称或嵌套映射，例如 dns.question.name.

请参阅导出字段以获取FileBeat导出的所有字段的列表。

支持的条件：

• equals
• contains
• regexp
• range
• network
• has_fields
• or
• and
• not

equals

比较字段是否具有特定值。该条件仅接受整数或字符串值。

例如，以下条件检查HTTP事务的响应码是否为200：

equals:
  http.response.code: 200

contains

检查值是否是字段的一部分。该字段可以是字符串或字符串数组。该条件仅接受字符串值。

例如，以下条件检查错误是否是事务状态的一部分：

contains:
  status: "Specific error"

regexp

根据正则表达式检查字段。该条件仅接受字符串。

例如，以下条件检查进程名称是否以foo开头：

regexp:
  system.process.name: "^foo.*"

range

检查字段是否在某个值范围内。该条件支持lte、lte、gt和gte。该条件仅接受整数或浮点值。

例如，以下条件通过将http.response.code字段与400进行比较来检查失败的HTTP事务。

range:
  http.response.code:
    gte: 400

这也可以写成：

range:
  http.response.code.gte: 400

以下条件检查CPU使用率（以百分比为单位）的值是否介于0.5和0.8之间。

range:
  system.cpu.user.pct.gte: 0.5
  system.cpu.user.pct.lt: 0.8

network

检查该字段是否在某个IP网络范围内。支持IPv4和IPv6地址。网络范围可以使用CIDR表示法指定，如“192.0.2.0/24”或“2001:db8::/32”，或者使用以下命名：

• loopback - 匹配回环地址 in the range of 127.0.0.0/8 or ::1/128.
• unicast - 匹配单播地址 defined in RFC 1122, RFC 4632, and RFC 4291，但排除IPv4广播地址 (255.255.255.255). 这包括私有地址范围。
• multicast - 匹配多播地址。
• interface_local_multicast - 匹配IPv6接口本地多播地址。
• link_local_unicast - 匹配链接本地单播地址。
• link_local_multicast - 匹配链接本地多播地址。
• private - 匹配RFC 1918（IPv4）和RFC 4193（IPv6）中定义的私有地址范围。
• public - Matches addresses that are not loopback, unspecified, IPv4 broadcast, link local unicast, link local multicast, interface local multicast, or private.
• unspecified - 匹配未指定的地址（IPv4地址0.0.0.0或IPv6地址::）。

如果source.ip值在私有地址空间内，则以下条件返回true。

network:
  source.ip: private

如果destination.ip值在192.168.1.0-192.168.1.255的IPv4范围内，则此条件返回true。

network:
  destination.ip: '192.168.1.0/24'

当destination.ip在任何给定子网内时，此条件返回true。

network:
  destination.ip: ['192.168.1.0/24', '10.0.0.0/8', loopback]

has_fields

查事件中是否存在所有给定字段。表示字段名称的字符串值列表。

例如，以下条件检查事件中是否存在http.response.code字段。

has_fields: ['http.response.code']

or

or运算符接收条件列表。

or:
  - <condition1>
  - <condition2>
  - <condition3>
  ...

例如，要配置条件http.response.code=304 OR http.response.code=404：

or:
  - equals:
      http.response.code: 304
  - equals:
      http.response.code: 404

and

and运算符接收条件列表。

and:
  - <condition1>
  - <condition2>
  - <condition3>
  ...

例如，要配置条件http.response.code=200 AND status=OK：

and:
  - equals:
      http.response.code: 200
  - equals:
      status: OK

要配置像<condition1> OR <condition2> AND <condition3>这样的条件：

or:
  - <condition1>
  - and:
    - <condition2>
    - <condition3>

not

not运算符接收到否定的条件。

not:
  <condition>

例如，要配置条件NOT status = OK：

not:
  equals:
    status: OK