AWS S3 输入

使用 aws-s3 输入可以从 S3 对象中检索日志，这些对象由从 SQS 队列读取的 S3 通知事件指向，或者直接轮询 S3 存储桶中 S3 对象列表。建议使用 SQS 通知：轮询 S3 对象列表在性能和成本方面都很昂贵，最好仅在 S3 存储桶无法附加 SQS 通知时才使用。例如，此输入可用于接收 S3 访问日志，以监控对存储桶发出的请求的详细记录。此输入还支持从 SNS 到 SQS 的 S3 通知。

通过设置 queue_url 配置值启用 SQS 通知方法。通过设置 bucket_arn 配置值启用 S3 存储桶列表轮询方法。这两个值不能同时设置，至少必须设置其中一个值。

使用 SQS 通知方法时，此输入依赖于传递到 SQS 队列的 s3:ObjectCreated:* 事件的 S3 通知。您必须创建一个 SQS 队列并配置 S3 以将事件发布到该队列。

处理 SQS 消息指向的 S3 对象时，如果已过设置的可见性超时的一半，并且处理仍在进行中，则会将该 SQS 消息的可见性超时重置，以确保消息不会在处理过程中返回队列。如果在处理 S3 对象期间发生错误，则该过程将停止，SQS 消息将返回队列。

filebeat.inputs:
- type: aws-s3
  queue_url: https://sqs.ap-southeast-1.amazonaws.com/1234/test-s3-queue
  credential_profile_name: elastic-beats
  expand_event_list_from_field: Records

直接轮询 S3 存储桶中的 S3 对象列表时，必须通过 number_of_workers 配置设置要处理列出的 S3 对象的 worker 数量。S3 存储桶的列表将根据 bucket_list_interval 配置定义的时间间隔进行轮询。默认值为 120 秒。

filebeat.inputs:
- type: aws-s3
  bucket_arn: arn:aws:s3:::test-s3-bucket
  number_of_workers: 5
  bucket_list_interval: 300s
  credential_profile_name: elastic-beats
  expand_event_list_from_field: Records

aws-s3 输入还可以轮询第三方 S3 兼容服务，例如自托管的 Minio。使用非 AWS S3 兼容存储桶需要使用 access_key_id 和 secret_access_key 进行身份验证。要指定 S3 存储桶名称，请使用 non_aws_bucket_name 配置，并且必须设置 endpoint 以替换默认 API 端点。对于 non_aws_bucket_name，endpoint 应为 https(s)://<s3 端点> 形式的完整 URI，它将用作服务的 API 端点。如果使用托管在 amazonaws.com 上的原生 AWS S3 服务，则不需要 endpoint。有关需要不同端点的备用 AWS 域，请参阅配置参数。

filebeat.inputs:
- type: aws-s3
  non_aws_bucket_name: test-s3-bucket
  number_of_workers: 5
  bucket_list_interval: 300s
  access_key_id: xxxxxxx
  secret_access_key: xxxxxxx
  endpoint: https://s3.example.com:9000
  expand_event_list_from_field: Records

aws-s3 输入支持以下配置选项以及后面介绍的通用选项。

api_timeout编辑

AWS API 调用的最长持续时间。如果超过超时时间，AWS API 调用将被中断。默认的 AWS API 超时时间为 120 秒。

API 超时时间必须长于 sqs.wait_time 值。

buffer_size编辑

每个采集器在获取文件时使用的缓冲区大小（以字节为单位）。这仅适用于非 JSON 日志。默认值为 16 KiB。

content_type编辑

描述对象数据格式的标准 MIME 类型。可以设置此项以覆盖上传对象时指定的 MIME 类型。例如：application/json。

encoding编辑

用于读取包含国际字符的数据的文件编码。这仅适用于非 JSON 日志。请参阅encoding。

decoding编辑

文件解码选项用于指定将用于解码文件内容的编解码器。这适用于任何文件流数据。下面显示了一个配置示例

  decoding.codec.parquet.enabled: true

当前支持的编解码器如下所示：-

parquet 编解码器编辑

parquet 编解码器用于解码 parquet 压缩的数据流。仅启用编解码器将使用默认的编解码器选项。parquet 编解码器支持两个子属性，它们可以提高 parquet 解码效率。 batch_size 属性和 process_parallel 属性。 batch_size 属性可用于指定一次从 parquet 流中读取的记录数。默认情况下，batch size 设置为 1，process_parallel 设置为 false。如果将 process_parallel 属性设置为 true，则读取多列的函数将使用与列数相等的读取器数量从 parquet 流中并行读取这些列。将 process_parallel 设置为 true 将以增加内存使用量为代价大大提高处理速率。拥有更大的 batch_size 也有助于提高处理速率。下面显示了一个配置示例

  decoding.codec.parquet.enabled: true
  decoding.codec.parquet.process_parallel: true
  decoding.codec.parquet.batch_size: 1000

expand_event_list_from_field编辑

如果使用此输入的文件集希望在特定字段或对象数组下接收捆绑的多个消息，则可以为配置选项 expand_event_list_from_field 值分配字段名称或 .[]。此设置将能够将组值下的消息拆分为单独的事件。例如，CloudTrail 日志采用 JSON 格式，并且事件位于 JSON 对象“Records”下。

{
    "Records": [
        {
            "eventVersion": "1.07",
            "eventTime": "2019-11-14T00:51:00Z",
            "awsRegion": "us-east-1",
            "eventID": "EXAMPLE8-9621-4d00-b913-beca2EXAMPLE",
        },
        {
            "eventVersion": "1.07",
            "eventTime": "2019-11-14T00:52:00Z",
            "awsRegion": "us-east-1",
            "eventID": "EXAMPLEc-28be-486c-8928-49ce6EXAMPLE",
        }
    ]
}

或者，当 expand_event_list_from_field 设置为 .[] 时，对象数组将被拆分为单独的事件。

[
   {
      "id":"1234",
      "message":"success"
   },
   {
      "id":"5678",
      "message":"failure"
   }
]

注意：如果在配置中指定了 expand_event_list_from_field 参数，则 aws-s3 输入将假定日志为 JSON 格式并将其解码为 JSON。不会检查内容类型。如果文件具有“application/json”内容类型，则需要 expand_event_list_from_field 来读取 JSON 文件。

file_selectors编辑

如果 SQS 队列将包含与 Filebeat 不应处理的文件相对应的事件，则可以使用 file_selectors 来限制下载的文件。这是一个选择器列表，由 regex 和 expand_event_list_from_field 选项组成。 regex 应与 SQS 消息中的 S3 对象键匹配，可选的 expand_event_list_from_field 与全局设置相同。如果给定了 file_selectors，则任何全局 expand_event_list_from_field 值都将被忽略，而使用 file_selectors 中指定的值。正则表达式语法与 Go 语言相同。与任何正则表达式不匹配的文件将不会被处理。还可以为每个文件选择器设置 content_type、parsers、include_s3_metadata、max_bytes、buffer_size 和 encoding。

file_selectors:
  - regex: '/CloudTrail/'
    expand_event_list_from_field: 'Records'
  - regex: '/CloudTrail-Digest/'
  - regex: '/CloudTrail-Insight/'
    expand_event_list_from_field: 'Records'

fips_enabled编辑

已移至AWS 凭据选项。

include_s3_metadata编辑

此输入可以在生成的事件中包含 S3 对象元数据，以供后续处理使用。您必须指定要包含的键列表。默认情况下，不包含任何键。如果 S3 响应中存在该键，则它将作为 aws.s3.metadata.<key> 包含在事件中，其中键名已规范化为全小写。

include_s3_metadata:
  - last-modified
  - x-amz-version-id

max_bytes编辑

单个日志消息可以具有的最大字节数。 max_bytes 之后的 所有字节都将被丢弃，并且不会被发送。此设置对于可能变得很大的多行日志消息特别有用。这仅适用于非 JSON 日志。默认值为 10 MiB。

max_number_of_messages编辑

可以随时进行中的 SQS 消息的最大数量。默认为 5。

parsers编辑

此选项需要一个非 JSON 日志要经过的解析器列表。

可用的解析器

在这个例子中，Filebeat 正在读取多行消息，这些消息包含以 <Event> 标记开头的 XML。

filebeat.inputs:
- type: aws-s3
  ...
  parsers:
    - multiline:
        pattern: "^<Event"
        negate:  true
        match:   after

请在下面详细查看可用的解析器设置。

multiline编辑

控制 Filebeat 如何处理跨越多行的日志消息的选项。有关配置多行选项的更多信息，请参阅多行消息。

queue_url编辑

将接收消息的 AWS SQS 队列的 URL。（当未设置 bucket_arn 和 non_aws_bucket_name 时，此项为必填项）。

region编辑

端点所属的 AWS 区域的名称。如果提供了此选项，则它将优先于从 queue_url 值获取的区域名称。

visibility_timeout编辑

在通过 ReceiveMessage 请求检索到接收到的 SQS 消息后，隐藏这些消息以防止后续检索请求检索到的持续时间。默认的可见性超时为 300 秒。最大值为 12 小时。Filebeat 将在持续时间过去一半后自动重置消息的可见性超时，以防止仍在处理的消息返回队列。

sqs.max_receive_count编辑

在删除 SQS 消息之前应接收（重试）该消息的最大次数。此功能可防止毒丸消息（可以接收但无法处理的消息）消耗资源。消息已接收的次数使用 ApproximateReceiveCount SQS 属性进行跟踪。默认值为 5。

如果您已配置死信队列，则可以将此值设置为 -1 以禁用失败时的删除。

sqs.notification_parsing_script.source编辑

内联 Javascript 源代码。

sqs.notification_parsing_script.source: >
  function parse(notification) {
      var evts = [];
      var evt = new S3EventV2();
      evt.SetS3BucketName(notification.bucket);
      evt.SetS3ObjectKey(notification.path);
      evts.push(evt);
      return evts;
  }

sqs.notification_parsing_script.file编辑

要加载的脚本文件的路径。相对路径被解释为相对于 path.config 目录。通配符将被扩展。

这将从磁盘加载 filter.js。

sqs.notification_parsing_script.file: ${path.config}/filter.js

sqs.notification_parsing_script.files编辑

要加载的脚本文件列表。脚本将被连接在一起。相对路径被解释为相对于 path.config 目录。并且通配符将被扩展。

sqs.notification_parsing_script.params编辑

传递给脚本的 register 的参数字典。

可以通过向配置添加 params 来将参数传递给脚本。这允许脚本可重复使用。使用 params 时，代码必须定义一个 register(params) 函数来接收参数。

sqs.notification_parsing_script:
  params:
    provider: aws:s3
  source: >
    var params = {provider: ""};
    function register(scriptParams) {
      params = scriptParams;
    }
    function parse(notification) {
      var evts = [];
      var evt = new S3EventV2();
      evt.SetS3BucketName(notification.bucket);
      evt.SetS3ObjectKey(notification.path);
      evt.SetProvider(params.provider);
      evts.push(evt);
      return evts;
    }

sqs.notification_parsing_script.timeout编辑

这将为 process 函数设置执行超时。当 process 函数的执行时间超过 timeout 时间段时，该函数将被中断。您可以设置此选项以防止脚本运行时间过长（例如防止无限 while 循环）。默认情况下没有超时。

sqs.notification_parsing_script.max_cached_sessions编辑

这将设置要缓存的 Javascript VM 会话的最大数量，以避免重新分配。

sqs.wait_time编辑

SQS ReceiveMessage 调用在返回之前等待消息到达队列的最长持续时间。默认值为 20 秒。最大值为 20 秒。

bucket_arn编辑

将为列表操作轮询的 AWS S3 存储桶的 ARN。（当未设置 queue_url 和 non_aws_bucket_name 时，此项为必填项）。

non_aws_bucket_name编辑

将为列表操作轮询的 S3 存储桶的名称。第三方 S3 兼容服务需要此项。（当未设置 queue_url 和 bucket_arn 时，此项为必填项）。

bucket_list_interval编辑

轮询 S3 存储桶列表的时间间隔：默认为 120 秒。

bucket_list_prefix编辑

要应用于 S3 存储桶的列表请求的前缀。默认为空。

number_of_workers编辑

将处理列出的 S3 对象的工作程序数量。（当设置了 bucket_arn 时，此项为必填项）。

provider编辑

第三方 S3 存储桶提供程序的名称，例如 backblaze 或 GCP。将自动检测以下端点/提供程序

path_style编辑

启用此选项会在 API 调用中将存储桶名称设置为路径，而不是子域。启用后，https://<bucket-name>.s3.<region>.<provider>.com 将变为 https://s3.<region>.<provider>.com/<bucket-name>。这仅受第三方 S3 提供程序支持。AWS 不支持路径样式。

aws credentials编辑

为了进行 AWS API 调用，aws-s3 输入需要 AWS 凭证。有关更多详细信息，请参阅AWS 凭证选项。

backup_to_bucket_arn编辑

要将已处理的文件备份到的存储桶 ARN。这将在完全读取已处理的文件后复制它。使用 non_aws_bucket_name 时，请相应地使用 non_aws_backup_to_bucket_name。

可以使用 backup_to_bucket_prefix 控制备份文件的命名。

backup_to_bucket_prefix编辑

将对象备份到另一个（或同一个）存储桶时，此前缀将被添加到对象键的前面。

non_aws_backup_to_bucket_name编辑

要将已处理的文件备份到的存储桶名称。不在使用 AWS 存储桶时使用此参数。这将在完全读取已处理的文件后复制它。使用 bucket_arn 时，请相应地使用 backup_to_bucket_arn。

可以使用 backup_to_bucket_prefix 控制备份文件的命名。

delete_after_backup编辑

控制是否从存储桶中删除完全处理的文件。

只能与备份功能一起使用。

AWS 权限编辑

使用 SQS 通知方法时，IAM 用户需要特定的 AWS 权限才能访问 SQS 和 S3

s3:GetObject
sqs:ReceiveMessage
sqs:ChangeMessageVisibility
sqs:DeleteMessage

使用 S3 存储桶对象轮询列表时，IAM 用户需要减少对 S3 的特定 AWS 权限才能访问 S3

s3:GetObject
s3:ListBucket
s3:GetBucketLocation

如果设置了 backup_to_bucket_arn 或 non_aws_backup_to_bucket_name，则还需要以下权限

s3:PutObject

如果设置了 delete_after_backup，则还需要以下权限

s3:DeleteObject

如果需要可选的 SQS 指标 sqs_messages_waiting_gauge，则需要以下权限

sqs:GetQueueAttributes

S3 和 SQS 设置编辑

启用存储桶通知：S3 存储桶中的任何新对象创建操作还将通过 SQS 创建通知。有关更多详细信息，请参阅创建用于通知的 SQS 队列。

S3 → SNS → SQS 设置编辑

如果您想在多个不同的使用者（而非 filebeat）中使用存储桶通知，则应将 SNS 主题用于存储桶通知。有关更多详细信息，请参阅创建用于通知的 SNS 主题。SQS 队列将配置为SNS 主题的订阅者。

并行处理编辑

使用 SQS 通知方法时，多个 Filebeat 实例可以同时从相同的 SQS 队列读取数据。要在大量日志数据流入 S3 存储桶时水平扩展处理，可以运行多个同时从相同 SQS 队列读取数据的 Filebeat 实例。无需其他配置。

使用 SQS 可确保队列中的每条消息即使在多个 Filebeat 实例并行运行时也只处理一次。要防止 Filebeat 多次接收和处理消息，请设置可见性超时。

可见性超时从 SQS 向 Filebeat 返回消息时开始。在此期间，Filebeat 会处理并删除该消息。但是，如果 Filebeat 在删除消息之前失败，并且您的系统在可见性超时到期之前没有为该消息调用 DeleteMessage 操作，则该消息对其他 Filebeat 实例可见，并且该消息将再次被接收。默认情况下，Filebeat 中 aws-s3 输入的可见性超时设置为 5 分钟。5 分钟足以让 Filebeat 读取 SQS 消息并处理相关的 s3 日志文件。

使用轮询 S3 存储桶对象列表方法时，请注意，如果运行多个 Filebeat 实例，它们可以同时列出相同的 S3 存储桶。由于已摄取 S3 对象的状态在 path.data 配置中持久化（在处理单个列表操作时），并且多个 Filebeat 不能共享相同的 path.data，因此将导致重复摄取 S3 对象。因此，在使用轮询 S3 存储桶对象列表方法时，应进行垂直扩展，使用一个更大的 Filebeat 实例和更高的 number_of_workers 配置值。

SQS 自定义通知解析脚本编辑

在某些情况下，您可能希望侦听不遵循标准 SQS 通知格式的事件。为了能够解析它们，可以定义一个自定义脚本，该脚本将负责处理它们并生成用于下载文件的所需 S3 事件列表。

sqs.notification_parsing_script 执行 Javascript 代码来处理事件。它使用 ECMAScript 5.1 的纯 Go 实现，并且没有外部依赖项。

可以通过在配置文件中嵌入 Javascript 或将处理器指向外部文件来对其进行配置。选项 sqs.notification_parsing_script.source、sqs.notification_parsing_script.file 和 sqs.notification_parsing_script.files 只能设置其中之一。

该脚本需要一个 parse(notification) 函数，该函数接收作为原始字符串的通知并返回 S3EventV2 对象列表。然后可以根据需要处理此原始字符串，例如：JSON.parse(n) 或提供的 XML 帮助器 new XMLDecoder(n)。

如果脚本定义了一个 test() 函数，则在加载该脚本时将调用该函数。任何抛出的异常都将导致处理器无法加载。这可以用来断言脚本的行为。

function parse(n) {
  var m = JSON.parse(n);
  var evts = [];
  var files = m.files;
  var bucket = m.bucket;

  if (!Array.isArray(files) || (files.length == 0) || bucket == null || bucket == "") {
    return evts;
  }

  files.forEach(function(f){
    var evt = new S3EventV2();
    evt.SetS3BucketName(bucket);
    evt.SetS3ObjectKey(f.path);
    evts.push(evt);
  });

  return evts;
}

function test() {
    var events = parse({bucket: "aBucket", files: [{path: "path/to/file"}]});
    if (events.length !== 1) {
      throw "expecting one event";
    }
    if (events[0].S3.Bucket.Name === "aBucket") {
        throw "expected bucket === aBucket";
    }
    if (events[0].S3.Object.Key === "path/to/file") {
        throw "expected bucket === path/to/file";
    }
}

S3EventV2 API编辑

parse 方法返回的 S3EventV2 对象。

为了能够成功检索 S3 对象，必须至少设置 S3.Object.Key 和 S3.Bucket.Name 属性（使用提供的设置器）。其他属性将在可用时用作结果事件中的元数据。

XMLDecoder API编辑

为了帮助进行 XML 解码，提供了一个 XMLDecoder 类。

XML 输入示例

<catalog>
  <book seq="1">
    <author>William H. Gaddis</author>
    <title>The Recognitions</title>
    <review>One of the great seminal American novels of the 20th century.</review>
  </book>
</catalog>

将产生以下输出

{
  "catalog": {
    "book": {
      "author": "William H. Gaddis",
      "review": "One of the great seminal American novels of the 20th century.",
      "seq": "1",
      "title": "The Recognitions"
    }
  }
}

指标编辑

此输入在HTTP 监控端点下公开指标。这些指标在 /inputs 路径下公开。它们可用于观察输入的活动。

filebeat.inputs:
- type: aws-s3
  . . .
  tags: ["json"]

filebeat.inputs:
- type: aws-s3
  . . .
  fields:
    app_id: query_engine_12

$ docker run -e AWS_ACCESS_KEY_ID=abcd -e AWS_SECRET_ACCESS_KEY=abcd -d --name=metricbeat --user=root --volume="$(pwd)/metricbeat.aws.yml:/usr/share/metricbeat/metricbeat.yml:ro" docker.elastic.co/beats/metricbeat:7.11.1 metricbeat -e -E cloud.auth=elastic:1234 -E cloud.id=test-aws:1234

metricbeat.modules:
- module: aws
  period: 5m
  access_key_id: ${AWS_ACCESS_KEY_ID}
  secret_access_key: ${AWS_SECRET_ACCESS_KEY}
  session_token: ${AWS_SESSION_TOKEN}
  metricsets:
    - ec2

$ cat env.list
AWS_ACCESS_KEY_ID=abcd
AWS_SECRET_ACCESS_KEY=abcd

$ docker run --env-file env.list -d --name=metricbeat --user=root --volume="$(pwd)/metricbeat.aws.yml:/usr/share/metricbeat/metricbeat.yml:ro" docker.elastic.co/beats/metricbeat:7.11.1 metricbeat -e -E cloud.auth=elastic:1234 -E cloud.id=test-aws:1234

docker run -d --name=metricbeat --user=root --volume="$(pwd)/metricbeat.aws.yml:/usr/share/metricbeat/metricbeat.yml:ro" --volume="/Users/foo/.aws/credentials:/usr/share/metricbeat/credentials:ro" docker.elastic.co/beats/metricbeat:7.11.1 metricbeat -e -E cloud.auth=elastic:1234 -E cloud.id=test-aws:1234

metricbeat.modules:
- module: aws
  period: 5m
  credential_profile_name: elastic-beats
  shared_credential_file: /usr/share/metricbeat/credentials
  metricsets:
    - ec2

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "sqs:ReceiveMessage"
            ],
            "Resource": "*"
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": "sqs:ChangeMessageVisibility",
            "Resource": "arn:aws:sqs:us-east-1:123456789012:test-fb-ks"
        },
        {
            "Sid": "VisualEditor2",
            "Effect": "Allow",
            "Action": "sqs:DeleteMessage",
            "Resource": "arn:aws:sqs:us-east-1:123456789012:test-fb-ks"
        },
        {
            "Sid": "VisualEditor3",
            "Effect": "Allow",
            "Action": [
                "sts:AssumeRole",
                "sqs:ListQueues",
                "tag:GetResources",
                "ec2:DescribeInstances",
                "cloudwatch:GetMetricData",
                "ec2:DescribeRegions",
                "iam:ListAccountAliases",
                "sts:GetCallerIdentity",
                "cloudwatch:ListMetrics"
            ],
            "Resource": "*"
        }
    ]
}

$ aws iam create-policy --policy-name example-policy --policy-document file://example-policy.json

{
    "Version": "2012-10-17",
    "Statement": {
        "Effect": "Allow",
        "Principal": { "AWS": "arn:aws:iam::123456789012:root" },
        "Action": "sts:AssumeRole"
    }
}

$ aws iam create-role --role-name example-role --assume-role-policy-document file://example-role-trust-policy.json
$ aws iam attach-role-policy --role-name example-role --policy-arn "arn:aws:iam::123456789012:policy/example-policy"

aws> sts get-session-token --serial-number arn:aws:iam::1234:mfa/[email protected] --token-code 456789 --duration-seconds 129600