opensearch

opensearch 源插件用于从 OpenSearch 集群、旧版 Elasticsearch 集群、Amazon OpenSearch Service 域或 Amazon OpenSearch Serverless 集合中读取索引。

该插件支持 OpenSearch 2.x 和 Elasticsearch 7.x。

用法

要使用具有最低所需设置的 opensearch 源，请将以下配置添加到您的 pipeline.yaml 文件中

opensearch-source-pipeline:
 source:
  opensearch:
    hosts: [ "https://:9200" ]
    username: "username"
    password: "password"
 ...

要使用包含所有配置设置（包括 indices、scheduling、search_options 和 connection）的 opensearch 源，请将以下示例添加到您的 pipeline.yaml 文件中

opensearch-source-pipeline:
  source:
    opensearch:
      hosts: [ "https://:9200" ]
      username: "username"
      password: "password"
      indices:
        include:
          - index_name_regex: "test-index-.*"
        exclude:
          - index_name_regex: "\..*"
      scheduling:
        interval: "PT1H"
        index_read_count: 2
        start_time: "2023-06-02T22:01:30.00Z"
      search_options:
        search_context_type: "none"
        batch_size: 1000
      connection:
        insecure: false
        cert: "/path/to/cert.crt"
  ...

Amazon OpenSearch 服务

opensearch 源可以通过传递一个具有域访问权限的 sts_role_arn 来配置 Amazon OpenSearch Service 域，如以下示例所示

opensearch-source-pipeline:
  source:
    opensearch:
      hosts: [ "https://search-my-domain-soopywaovobopgs8ywurr3utsu.us-east-1.es.amazonaws.com" ]
      aws:
        region: "us-east-1"
        sts_role_arn: "arn:aws:iam::123456789012:role/my-domain-role"
  ...

Amazon OpenSearch Serverless

opensearch 源可以通过将 serverless 选项设置为 true 来配置 Amazon OpenSearch Serverless，如以下示例所示

    - opensearch:
        hosts: [ 'https://1234567890abcdefghijkl.us-west-2.aoss.amazonaws.com' ]
        aws:
          sts_role_arn: 'arn:aws:iam::123456789012:role/my-domain-role'
          region: 'us-west-2'
          serverless: true

使用元数据

当 opensource 源从集群中的文档构建 OpenSearch Data Prepper 事件时，文档索引存储在 EventMetadata 中，其键为 opensearch-index，而 document_id 则存储在 EventMetadata 中，其键为 opensearch-document_id。这允许基于索引或 document_id 进行条件路由。以下示例管道配置将事件发送到 opensearch 接收器，并使用与目标集群中源集群相同的索引和 document_id

opensearch-migration-pipeline:
  source:
    opensearch:
      hosts: [ "https://source-cluster:9200" ]
      username: "username"
      password: "password"
  sink:
    - opensearch:
        hosts: [ "https://sink-cluster:9200" ]
        username: "username"
        password: "password"
        document_id: "${getMetadata(\"opensearch-document_id\")}"
        index: "${getMetadata(\"opensearch-index\"}"

配置选项

下表描述了您可以为 opensearch 源配置的选项。

Serverless 选项

以下选项可在 serverless_options 对象中使用。

调度

scheduling 配置允许用户根据 index_read_count 和重计数时间 interval 配置源中索引的重新处理方式。

例如，将 index_read_count 设置为 3，interval 设置为 1h 将导致所有索引被重新处理 3 次，每次间隔 1 小时。默认情况下，索引只会处理一次。

在 scheduling 配置下使用以下选项。

索引

以下选项帮助 opensearch 源使用正则表达式模式确定从源集群处理哪些索引。只有当索引匹配 include 设置下的一个 index_name_regex 模式且不匹配 exclude 设置下的任何模式时，才会对其进行处理。

在 include 和 exclude 选项下使用以下设置来指示索引的正则表达式模式。

search_options

在 search_options 配置下使用以下设置。

默认搜索行为

默认情况下，opensearch 源将查找集群版本和分布以确定要使用的 search_context_type。对于支持时间点的版本和分布，将使用 point_in_time。如果集群不支持 point_in_time，则将使用滚动。对于 Amazon OpenSearch Serverless 集合，将使用search_after，因为集合不支持 point_in_time 和 scroll。

连接

在 connection 配置下使用以下设置。

AWS

设置 aws 服务的身份验证时使用以下选项。

指标

opensearch 源包含以下指标。

计数器

• documentsProcessed：衡量 opensearch 源插件处理的文档总数。
• indicesProcessed：衡量 opensearch 源插件处理的索引总数。
• processingErrors：衡量 opensearch 源插件造成的索引处理错误总数。
• credentialsChanged：衡量 opensearch 源刷新基本凭证（用户名/密码）的次数。
• clientRefreshErrors：衡量由于 opensearch 源刷新基本凭证而生成新客户端时遇到的错误数量。

计时器

• indexProcessingTime：衡量 opensearch 源插件索引处理延迟，以秒为单位。

分布汇总

• bytesReceived：衡量 opensearch 源插件接收的传入文档的大小分布，以字节为单位。
• bytesProcessed：衡量 opensearch 源插件成功处理的传入文档的大小分布，以字节为单位。

OpenSearch 集群安全

为了使用 opensearch 源插件从 OpenSearch 集群拉取数据，您必须在管道配置中指定用户名和密码。以下 pipeline.yaml 示例文件演示了如何指定默认的管理员安全凭证

source:
  opensearch:
    username: "admin"
    password: "admin"
  ...

Amazon OpenSearch Service 域安全

opensearch 源插件可以从使用 AWS Identity and Access Management (IAM) 进行安全的Amazon OpenSearch Service 域拉取数据。该插件使用默认的 Amazon OpenSearch Service 凭证链。使用 AWS 命令行界面 (AWS CLI) 运行 aws configure 来设置您的凭证。

确保您配置的凭证具有所需的 IAM 权限。以下域访问策略显示了最低所需权限

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::<AccountId>:user/data-prepper-user"
      },
      "Action": "es:ESHttpGet",
      "Resource": [
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/_cat/indices",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/_search",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/_search/scroll",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/*/_search"
      ]
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::<AccountId>:user/data-prepper-user"
      },
      "Action": "es:ESHttpPost",
      "Resource": [
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/*/_search/point_in_time",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/*/_search/scroll"
      ]
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::<AccountId>:user/data-prepper-user"
      },
      "Action": "es:ESHttpDelete",
      "Resource": [
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/_search/point_in_time",
        "arn:aws:es:us-east-1:<AccountId>:domain/<domain-name>/_search/scroll"
      ]
    }
  ]
}

有关如何配置域访问策略的说明，请参阅 Amazon OpenSearch Service 文档中的基于资源的策略。

OpenSearch Serverless 集合安全

opensearch 源插件可以从Amazon OpenSearch Serverless 集合接收数据。

您无法从使用虚拟私有云 (VPC) 访问的集合中读取数据。集合必须可从公共网络访问。

创建管道角色

要使用 OpenSearch Serverless 集合安全，请创建一个 IAM 角色，管道将扮演该角色以从集合中读取数据。该角色必须具有以下最低权限

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "aoss:APIAccessAll"
            ],
            "Resource": "arn:aws:aoss:*:<AccountId>:collection/*"
        }
    ]
}

创建集合

接下来，创建具有以下设置的集合

• 对 OpenSearch 端点和 OpenSearch Dashboards 的公共网络访问。

• 以下数据访问策略，它授予管道角色所需的权限，如以下配置所示

  [
   {
      "Rules":[
         {
            "Resource":[
               "index/collection-name/*"
            ],
            "Permission":[
               "aoss:ReadDocument",
               "aoss:DescribeIndex"
            ],
            "ResourceType":"index"
         }
      ],
      "Principal":[
         "arn:aws:iam::<AccountId>:role/PipelineRole"
      ],
      "Description":"Pipeline role access"
   }
  ]
  

确保将 Principal 元素中的 Amazon Resource Name (ARN) 替换为您在上一步中创建的管道角色的 ARN。

有关如何创建集合的说明，请参阅 Amazon OpenSearch Service 文档中的创建集合。

创建管道

在您的 pipeline.yaml 文件中，将 OpenSearch Serverless 集合端点指定为 hosts 选项。此外，您必须将 serverless 选项设置为 true。在 sts_role_arn 选项中指定管道角色，如以下示例所示

opensearch-source-pipeline:
  source:
    opensearch:
      hosts: [ "https://<serverless-public-collection-endpoint>" ]
      aws:
        serverless: true
        sts_role_arn: "arn:aws:iam::<AccountId>:role/PipelineRole"
        region: "us-east-1"
  processor:
    - date:
        from_time_received: true
        destination: "@timestamp"
  sink:
    - stdout: