s3 源

s3 是一个源插件，用于从 Amazon Simple Storage Service (Amazon S3) 对象读取事件。您可以将源配置为使用 Amazon Simple Queue Service (Amazon SQS) 队列或扫描 S3 存储桶

• 要使用 Amazon SQS 通知，请在您的 S3 存储桶上配置 S3 事件通知。配置 Amazon SQS 后，s3 源将从 Amazon SQS 接收消息。当 SQS 消息指示已创建 S3 对象时，s3 源将加载 S3 对象，然后使用配置的 编解码器解析它们。
• 要使用 S3 存储桶，请配置 s3 源以使用 Amazon S3 Select 而非 OpenSearch Data Prepper 来解析 S3 对象。

IAM 权限

为了使用 s3 源，请配置您的 AWS Identity and Access Management (IAM) 权限，以授予 Data Prepper 访问 Amazon S3 的权限。您可以使用类似于以下 JSON 配置的配置

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "s3-access",
            "Effect": "Allow",
            "Action": [
              "s3:GetObject",
              "s3:ListBucket",
              "s3:DeleteObject"
            ],
            "Resource": "arn:aws:s3:::<YOUR-BUCKET>/*"
        },
        {
            "Sid": "sqs-access",
            "Effect": "Allow",
            "Action": [
                "sqs:ChangeMessageVisibility",
                "sqs:DeleteMessage",
                "sqs:ReceiveMessage"
            ],
            "Resource": "arn:aws:sqs:<YOUR-REGION>:<123456789012>:<YOUR-SQS-QUEUE>"
        },
        {
            "Sid": "kms-access",
            "Effect": "Allow",
            "Action": "kms:Decrypt",
            "Resource": "arn:aws:kms:<YOUR-REGION>:<123456789012>:key/<YOUR-KMS-KEY>"
        }
    ]
}

如果您的 S3 对象或 Amazon SQS 队列不使用 AWS Key Management Service (AWS KMS)，请移除 kms:Decrypt 权限。

如果您未启用 visibility_duplication_protection，则可以从 SQS 队列的访问权限中移除 sqs:ChangeMessageVisibility 权限。

跨账户 S3 访问

当 Data Prepper 从 S3 存储桶获取数据时，它使用存储桶所有者条件验证存储桶的所有权。默认情况下，Data Prepper 期望 S3 存储桶由拥有相关 SQS 队列的相同账户拥有。如果没有提供 SQS，Data Prepper 将使用 aws 配置中的 Amazon Resource Name (ARN) 角色。

如果您计划从多个 S3 存储桶摄取数据，但每个存储桶都与不同的 S3 账户关联，则需要根据以下条件配置 Data Prepper 以检查跨账户 S3 访问

• 如果您想要数据的 S3 存储桶都属于 SQS 队列账户以外的账户，请将 default_bucket_owner 设置为存储桶账户持有者的账户 ID。
• 如果您的 S3 存储桶位于多个账户中，请使用 bucket_owners 映射。

在以下示例中，SQS 队列由账户 000000000000 拥有。SQS 队列包含来自两个 S3 存储桶的数据：my-bucket-01 和 my-bucket-02。由于 my-bucket-01 由 123456789012 拥有，而 my-bucket-02 由 999999999999 拥有，因此 bucket_owners 映射使用其账户 ID 调用这两个存储桶所有者，如以下配置所示

s3:
  sqs:
      queue_url: "https://sqs.us-east-1.amazonaws.com/000000000000/MyQueue"
  bucket_owners:
    my-bucket-01: 123456789012
    my-bucket-02: 999999999999

您可以同时使用 bucket_owners 和 default_bucket_owner。

配置

您可以使用以下选项配置 s3 源。

sqs

以下参数允许您在 s3 源插件中配置 Amazon SQS 的使用。

aws

编解码器

编解码器决定 s3 源如何解析每个 Amazon S3 对象。为了提高效率和性能，您可以将编解码器组合与某些处理器一起使用。

newline 编解码器

newline 编解码器将每一单行解析为单个日志事件。这对于大多数应用程序日志是理想的，因为每个事件都按单行解析。它也适用于每行具有独立 JSON 对象的 S3 对象，这在使用 parse_json 处理器解析每行时非常匹配。

使用以下选项配置 newline 编解码器。

json 编解码器

json 编解码器将每个 S3 对象解析为 JSON 数组中的单个 JSON 对象，然后为数组中的每个对象创建一个 Data Prepper 日志事件。

csv 编解码器

csv 编解码器解析逗号分隔值 (CSV) 格式的对象，每行生成一个 Data Prepper 日志事件。使用以下选项配置 csv 编解码器。

将 s3_select 与 s3 源一起使用

配置 s3_select 以解析 Amazon S3 对象时，请使用以下选项

csv

结合 s3_select 的 csv 配置使用以下选项，以确定解析后的 CSV 文件应如何格式化。

这些选项直接映射到 S3 Select CSVInput 数据类型中可用的选项。

json

结合 s3_select 的 json 使用以下选项，以确定 S3 Select 如何处理 JSON 文件。

将 scan 与 s3 源一起使用

以下参数允许您扫描 S3 对象。所有选项都可以在存储桶级别进行配置。

扫描存储桶

您可以在 bucket 设置映射中配置以下选项。

filter

在 filter 配置中使用以下选项。

scheduling

指标

s3 源包括以下指标

计数器

• s3ObjectsFailed：s3 源未能读取的 S3 对象数量。
• s3ObjectsNotFound：s3 源因 S3“未找到”错误而未能读取的 S3 对象数量。这些也计入 s3ObjectsFailed。
• s3ObjectsAccessDenied：s3 源因“访问被拒绝”或“禁止”错误而未能读取的 S3 对象数量。这些也计入 s3ObjectsFailed。
• s3ObjectsSucceeded：s3 源成功读取的 S3 对象数量。
• s3ObjectNoRecordsFound：s3 源向缓冲区添加了 0 条记录的 S3 对象数量。
• s3ObjectsDeleted：s3 源删除的 S3 对象数量。
• s3ObjectsDeleteFailed：s3 源未能删除的 S3 对象数量。
• s3ObjectsEmpty：被视为为空的 S3 对象数量，因为它们的大小为 0。s3 源将跳过这些对象。
• sqsMessagesReceived：s3 源从队列接收的 Amazon SQS 消息数量。
• sqsMessagesDeleted：s3 源从队列删除的 Amazon SQS 消息数量。
• sqsMessagesFailed：s3 源未能解析的 Amazon SQS 消息数量。
• sqsMessagesDeleteFailed – s3 源未能从 SQS 队列删除的 SQS 消息数量。
• sqsVisibilityTimeoutChangedCount：s3 源更改 SQS 消息可见性超时的次数。这包括同一消息上的多次可见性超时更改。
• sqsVisibilityTimeoutChangeFailedCount：s3 源未能更改 SQS 消息可见性超时的次数。这包括同一消息上的多次可见性超时更改失败。
• acknowledgementSetCallbackCounter：s3 源从 Data Prepper 接收到确认的次数。

计时器

• s3ObjectReadTimeElapsed：测量 s3 源执行获取 S3 对象请求、解析并写入事件到缓冲区所需的时间量。
• sqsMessageDelay：测量从 S3 创建对象到其完全解析所经过的时间。

分布汇总

• s3ObjectSizeBytes：测量 S3 对象的大小，由 S3 Content-Length 报告。对于压缩对象，这是压缩后的大小。
• s3ObjectProcessedBytes：测量 s3 源为给定对象处理的字节数。对于压缩对象，这是未压缩的大小。
• s3ObjectsEvents：测量 S3 对象生成的事件（有时称为记录）的数量。

示例：使用 SQS 的未压缩日志

以下 pipeline.yaml 文件显示了读取未压缩的换行符分隔日志的最小配置

source:
  s3:
    notification_type: sqs
    codec:
      newline:
    compression: none
    sqs:
      queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
    aws:
      region: "us-east-1"
      sts_role_arn: "arn:aws:iam::123456789012:role/Data-Prepper"

示例：使用扫描的未压缩日志

以下 pipeline.yaml 文件显示了扫描具有未压缩的换行符分隔日志的对象的最小配置

source:
  s3:
    codec:
      newline:
    compression: none
    aws:
      region: "us-east-1"
      sts_role_arn: "arn:aws:iam::123456789012:role/Data-Prepper"
    scan:
      start_time: 2023-01-01T00:00:00
      range: "P365D"
      buckets:
        - bucket:
            name: "s3-scan-test"
            filter:
              exclude_suffix:
                - "*.log"