Elastic 集成过滤器插件

编辑

Elastic 集成过滤器插件编辑

  • 插件版本:v0.1.11
  • 发布日期:2024-07-02
  • 变更日志

有关其他版本,请参阅 版本化插件文档.

获取帮助编辑

有关插件的问题,请在 Discuss 论坛中发布主题。对于错误或功能请求,请在 Github 中创建问题。有关 Elastic 支持的插件列表,请参阅 Elastic 支持矩阵.

描述编辑

使用此过滤器来处理 Logstash 中由 Elasticsearch Ingest Node 提供支持的 Elastic 集成。

当您将此过滤器配置为指向 Elasticsearch 集群时,它会使用显式定义的 pipeline_name 或自动检测事件的数据流及其默认管道来检测每个事件应执行哪个摄取管道(如果有)。

然后,它从 Elasticsearch 加载该管道的定义,并在 Logstash 中运行该管道,而不会将事件传输到 Elasticsearch。成功由其摄取管道处理的事件将具有 [@metadata][target_ingest_pipeline] 设置为 _none,以便 Logstash 管道中的任何下游 Elasticsearch 输出将避免在 Elasticsearch 中再次运行事件的默认管道。

某些多管道配置(例如通过 http(s) 的 logstash-to-logstash)不维护 [@metadata] 字段的状态。在这些设置中,您可能需要使用 pipeline => "_none" 显式配置下游管道的 Elasticsearch 输出,以避免重新运行默认管道。

失败摄取管道处理的事件将被标记为 _ingest_pipeline_failure,并且它们的 [@metadata][_ingest_pipeline_failure] 将以键值映射的形式填充详细信息。

此插件需要最低 Java 17 和 Logstash 8.7.0 版本。

filter-elastic_integrationoutput-elasticsearch 一起使用编辑

Elastic 集成旨在与 数据流ECS 兼容 输出一起使用。请确保在 output-elasticsearch 插件中启用了这些功能。

有关其他设置,请查看 output-elasticsearch 插件 文档。

最小配置编辑

您需要配置此插件以连接到 Elasticsearch,并且可能还需要提供本地 GeoIp 数据库。

filter {
  elastic_integration {
    cloud_id   => "YOUR_CLOUD_ID_HERE"
    cloud_auth => "YOUR_CLOUD_AUTH_HERE"
    geoip_database_directory => "/etc/your/geoip-databases"
  }
}

继续阅读以获取配置指南,或跳转到 完整的配置选项列表.

连接到 Elasticsearch编辑

此插件与 Elasticsearch 通信以识别应为给定事件运行哪个摄取管道,以及检索摄取管道定义本身。您必须配置此插件以使用以下一项指向 Elasticsearch

  • 云 ID(请参阅 cloud_id
  • 一个或多个主机 URL 列表(请参阅 hosts

除非您显式配置此插件,否则通信将通过 SSL 安全进行。

您可能需要配置此插件如何建立对响应服务器的信任,并且可能需要配置此插件如何呈现其自身身份或凭据。

SSL 信任配置编辑

在通过 SSL 通信时,此插件使用系统信任存储完全验证 Elasticsearch 提供的身份证明。您可以使用以下一项提供备用信任来源

您还可以配置身份证明的哪些方面将被验证(请参阅 ssl_verification_mode)。

SSL 身份配置编辑

在通过 SSL 通信时,您还可以配置此插件以使用以下一项向其连接到的 Elasticsearch 集群呈现基于证书的身份证明

请求身份编辑

您可以配置此插件以通过以下几种方式之一向 Elasticsearch 呈现身份验证凭据

您的请求凭据仅与传递它们的连接一样安全。它们本身不提供隐私或保密性,并且当 SSL 被禁用时,攻击者可以轻松地恢复它们。

最低所需权限编辑

此插件与 Elasticsearch 通信以将事件解析为管道定义,并且需要使用具有适当权限的凭据进行配置,以从相关 API 读取数据。在启动阶段,此插件确认当前用户具有足够的权限,包括

权限名称 描述

monitor

集群操作(如集群运行状况或状态)的只读权限。当插件检查 Elasticsearch 许可证时,插件需要它。

read_pipeline

对摄取管道的只读获取和模拟访问。当插件读取 Elasticsearch 摄取管道定义时,需要它。

manage_index_templates

对索引模板的所有操作权限。当插件根据事件数据流名称解析默认管道时,需要它。

当此插件连接到已禁用安全功能的 Elasticsearch 集群或用户未提供凭据时,此插件无法确定匿名用户是否具有所需的权限。插件以不安全模式启动,并出现运行时错误,表明 API 权限不足,并阻止事件由摄取管道处理。

为了避免这些问题,请设置用户身份验证并确保 Elasticsearch 中的安全性已启用(默认设置)。

支持的摄取处理器编辑

此过滤器可以运行完全由支持的处理器子集组成的 Elasticsearch Ingest Node 管道。它可以在适用时访问 Painless 和 Mustache 脚本引擎

来源 处理器 注意事项

摄取通用

append

bytes

communityid

convert

csv

date

dateindexname

dissect

dotexpander

drop

fail

fingerprint

foreach

grok

gsub

htmlstrip

join

json

keyvalue

lowercase

networkdirection

pipeline

解析的管道必须完全由支持的处理器组成

registereddomain

remove

rename

reroute

script

lang 必须为 painless(默认值)

set

sort

split

trim

uppercase

uri_parts

urldecode

user_agent

不支持侧载自定义正则表达式文件;处理器将使用 Elasticsearch 处理器定义 中指定的默认用户代理定义

Redact

redact

GeoIp

geoip

需要 MaxMind GeoIP2 数据库,这些数据库可以由 Logstash 的 Geoip 数据库管理使用 geoip_database_directory 进行配置

字段映射编辑

在执行过程中,摄取管道使用 Logstash 事件的临时可变视图,称为摄取文档。此视图包含来自事件的所有结构化字段,并进行最少的类型转换。

它还包含摄取管道处理器所需的额外元数据字段。

  • _version:一个long值整数,等效于事件的@version,或一个合理的默认值1
  • _ingest.timestamp:一个ZonedDateTime,等效于事件的@timestamp字段。

执行完成后,事件将被清理,以确保 Logstash 保留的字段具有预期的形状,并为任何缺少的必需字段提供合理的默认值。当摄取管道将保留字段设置为无法强制转换的值时,该值将在事件的备用位置提供,如下所述。

Logstash 字段 类型

@timestamp

时间戳

摄取文档的@timestampevent.created_ingest.timestamp_now字段的第一个可强制转换值;或当前时间戳。当摄取文档的@timestamp值无法强制转换时,它将在事件的_@timestamp字段中提供。

@version

字符串编码的整数

摄取文档的@version_version字段的第一个可强制转换值;或当前时间戳。当摄取文档的@version值无法强制转换时,它将在事件的_@version字段中提供。

@metadata

键值映射

摄取文档的@metadata;或空映射。当摄取文档的@metadata值无法强制转换时,它将在事件的_@metadata字段中提供。

tags

字符串或字符串列表

摄取文档的tags。当摄取文档的tags值无法强制转换时,它将在事件的_tags字段中提供。

此外,这些 Elasticsearch IngestDocument 元数据字段将在结果事件中提供,当且仅当它们在管道执行期间设置时。

Elasticsearch 文档元数据 Logstash 字段

_id

[@metadata][_ingest_document][id]

_index

[@metadata][_ingest_document][index]

_routing

[@metadata][_ingest_document][routing]

_version

[@metadata][_ingest_document][version]

_version_type

[@metadata][_ingest_document][version_type]

_ingest.timestamp

[@metadata][_ingest_document][timestamp]

解析管道定义编辑

此插件使用 Elasticsearch 将管道名称解析为其管道定义。当没有显式配置pipeline_name时,或者当管道使用 Reroute 处理器时,它还使用 Elasticsearch 建立数据流名称与其各自的默认管道名称的映射。

它使用命中/未命中缓存来避免为每个事件查询 Elasticsearch。它还努力在缓存映射过期之前更新这些缓存映射。结果是,当 Elasticsearch 响应时,此插件能够快速获取更改,而不会影响其自身的性能,并且它可以通过继续使用可能过时的映射或定义来承受 Elasticsearch 问题期间的停机。

为了实现这一点,映射的缓存时间最长为 24 小时,并且每 1 分钟重新加载一次缓存值,效果如下

  • 当重新加载的映射非空且与其已缓存的值相同时,其生存时间将重置,以确保后续事件可以继续使用已确认的未更改值。
  • 当重新加载的映射非空且与其先前缓存的值不同时,该条目将更新,以便后续事件将使用新值。
  • 当重新加载的映射为空时,先前非空的映射将被替换为新的空条目,以便后续事件将使用空值。
  • 当映射的重新加载失败时,此插件会发出日志警告,但现有缓存条目保持不变,并接近其过期时间。

Elastic Integration 过滤器配置选项编辑

此插件支持以下配置选项,以及后面描述的通用选项

api_key编辑

  • 值类型为密码
  • 此设置没有默认值。

用于将此插件身份验证到 Elasticsearch 的 API 密钥的编码形式。

cloud_auth编辑

  • 值类型为密码
  • 此设置没有默认值。

云身份验证字符串(“<username>:<password>” 格式)是 username/password 对的替代方案,可以从 Elastic Cloud 网页控制台获取。

cloud_id编辑

来自 Elastic Cloud 网页控制台的云 ID。

使用云 ID 连接时,与 Elasticsearch 的通信将使用 SSL 保护。

有关更多详细信息,请查看Logstash 到云文档

geoip_database_directory编辑

  • 值类型为路径
  • 此设置没有默认值。

在启用 Geoip 数据库管理的 Logstash 进程中运行时,使用 Geoip 处理器的集成默认情况下将使用托管的 Maxmind 数据库。使用托管数据库,即表示您接受并同意MaxMind EULA

您可以改为使用包含数据库文件的本地目录的路径配置此插件。

此插件将在提供的目录中发现所有具有.mmdb 后缀的常规文件,并通过其文件名使每个文件对集成管道中的 GeoIp 处理器可用。它期望找到的文件采用 MaxMind DB 格式,并具有以下数据库类型之一

  • AnonymousIp
  • ASN
  • City
  • Country
  • ConnectionType
  • Domain
  • Enterprise
  • Isp

大多数集成依赖于数据库的名称完全

  • GeoLite2-ASN.mmdb,
  • GeoLite2-City.mmdb
  • GeoLite2-Country.mmdb

hosts编辑

  • 值类型为uri 列表
  • 此设置没有默认值。
  • 约束

    • 当任何 URL 包含协议组件时,所有 URL 必须具有相同的协议。
    • https 协议主机使用 HTTPS,不能与ssl_enabled => false 结合使用。
    • http 协议主机使用不安全的 HTTP,不能与ssl_enabled => true 结合使用。
    • 当任何 URL 省略端口组件时,将使用默认的9200
    • 当任何 URL 包含路径组件时,所有 URL 必须具有相同的路径。

要连接的 Elasticsearch 主机列表(非空)。

示例

  • "127.0.0.1"
  • ["127.0.0.1:9200","127.0.0.2:9200"]
  • ["http://127.0.0.1"]
  • ["https://127.0.0.1:9200"]
  • ["https://127.0.0.1:9200/subpath"](如果使用代理在子路径上)

使用主机列表连接时,与 Elasticsearch 的通信将使用 SSL 保护,除非另有配置。

禁用 SSL 很危险

此插件的安全性依赖于 SSL 来避免泄露凭据,并避免运行非法的摄取管道定义。

有两种方法可以禁用 SSL

  • 提供http 协议主机的列表
  • 设置<<plugins-{type}s-{plugin}-ssl_enabled>>=>false

password编辑

  • 值类型为密码
  • 此设置没有默认值。
  • 当使用username 配置请求身份验证时,此项为必需项。

使用 HTTP 基本身份验证连接到 Elasticsearch 时使用的密码。

pipeline_name编辑

  • 值类型为字符串
  • 此设置没有默认值。
  • 如果存在,事件的初始管道不会从事件的数据流字段中自动检测。
  • 值可以是sprintf 样式 模板;如果任何引用的字段无法解析,则事件将不会路由到摄取管道。

ssl_certificate编辑

  • 值类型为路径
  • 此设置没有默认值。
  • 如果存在,ssl_keyssl_key_passphrase 也是必需的。
  • 不能与禁用 SSL 的配置结合使用。

PEM 编码证书或证书链的路径,用于将此插件标识到 Elasticsearch。

ssl_certificate_authoritiesedit

  • 值类型为 路径 列表
  • 此设置没有默认值。
  • 不能与禁用 SSL 的配置结合使用。
  • 不能与 `ssl_verification_mode⇒none` 结合使用。

一个或多个定义证书颁发机构的 PEM 格式文件。

此设置可用于 *覆盖* 系统信任存储,以验证 Elasticsearch 提供的 SSL 证书。

ssl_enablededit

  • 值类型为 布尔值
  • 此设置没有默认值。

除非以下情况,否则将启用与 Elasticsearch 的安全 SSL 通信

  • 它被 ssl_enabled => false 显式禁用;或者
  • 它通过提供 http 协议 hosts 隐式禁用。

指定 ssl_enabled => true 可以作为有用的冗余安全措施,以确保此插件无法配置为使用非 ssl 通信。

ssl_keyedit

  • 值类型为路径
  • 此设置没有默认值。
  • 当连接身份使用 ssl_certificate 配置时,这是必需的。
  • 不能与禁用 SSL 的配置结合使用。

PKCS8 格式的 SSL 证书密钥的路径。

ssl_keystore_passwordedit

  • 值类型为密码
  • 此设置没有默认值。
  • 当连接身份使用 ssl_keystore_path 配置时,这是必需的。
  • 不能与禁用 SSL 的配置结合使用。

ssl_keystore_path 的密码。

ssl_keystore_pathedit

  • 值类型为路径
  • 此设置没有默认值。
  • 如果存在,则还需要 ssl_keystore_password
  • 不能与禁用 SSL 的配置结合使用。

JKS 或 PKCS12 格式的密钥库的路径,用于向 Elasticsearch 标识此插件。

ssl_key_passphraseedit

  • 值类型为密码
  • 此设置没有默认值。
  • 当连接身份使用 ssl_certificate 配置时,这是必需的。
  • 不能与禁用 SSL 的配置结合使用。

ssl_key 的密码或口令。

ssl_truststore_pathedit

JKS 或 PKCS12 格式的密钥库的路径,其中包含受信任的证书。

此设置可用于 *覆盖* 系统信任存储,以验证 Elasticsearch 提供的 SSL 证书。

ssl_truststore_passwordedit

  • 值类型为密码
  • 此设置没有默认值。
  • 当连接信任使用 ssl_truststore_path 配置时,这是必需的。
  • 不能与禁用 SSL 的配置结合使用。

ssl_truststore_path 的密码。

ssl_verification_modeedit

  • 值类型为字符串
  • 此设置没有默认值。
  • 不能与禁用 SSL 的配置结合使用。

验证 Elasticsearch 提供的证书的级别。

默认情况下,Elasticsearch 提供的 SSL 证书将被完全验证。

  • 可用模式

    • none:不执行任何验证,隐式信任此插件连接到的任何服务器(不安全)
    • certificate:验证服务器提供的证书是由受信任的证书颁发机构签署的,并且服务器可以证明拥有其关联的私钥(不太安全)
    • full(默认):执行与 certificate 相同的验证,并验证提供的证书是否具有与我们尝试连接到的服务器匹配的身份声明(最安全)

usernameedit

  • 值类型为字符串
  • 此设置没有默认值。
  • 如果存在,则还需要 password

使用 HTTP 基本身份验证连接到 Elasticsearch 时的用户名。

 

通用选项edit

以下配置选项受所有过滤器插件支持

add_fieldedit

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

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

示例

    filter {
      elastic_integration {
        add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
      }
    }
    # You can also add multiple fields at once:
    filter {
      elastic_integration {
        add_field => {
          "foo_%{somefield}" => "Hello world, from %{host}"
          "new_field" => "new_static_value"
        }
      }
    }

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

add_tagedit

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

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

示例

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

如果事件具有字段 "somefield" == "hello",则此过滤器在成功时将添加标签 foo_hello(当然,第二个示例将添加 taggedy_tag 标签)。

enable_metricedit

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

idedit

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

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

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

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

periodic_flushedit

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

remove_fieldedit

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

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

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

如果事件具有字段 "somefield" == "hello",则此过滤器在成功时将删除名为 foo_hello 的字段(如果存在)。第二个示例将删除一个额外的非动态字段。

remove_tagedit

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

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

示例

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

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