s3 source

s3 is a source plugin that reads events from Amazon Simple Storage Service (Amazon S3) objects. It requires an Amazon Simple Queue Service (Amazon SQS) queue that receives S3 Event Notifications. After Amazon SQS is configured, the s3 source receives messages from Amazon SQS. When the SQS message indicates that an S3 object was created, the s3 source loads the S3 objects and then parses them using the configured codec. You can also configure the s3 source to use Amazon S3 Select instead of Data Prepper to parse S3 objects.

IAM permissions

In order to use the s3 source, configure your AWS Identity and Access Management (IAM) permissions to grant Data Prepper access to Amazon S3. You can use a configuration similar to the following JSON configuration:

{
    "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: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>"
        }
    ]
}

If your S3 objects or Amazon SQS queues do not use AWS Key Management Service (AWS KMS), remove the kms:Decrypt permission.

Cross-account S3 access

When Data Prepper fetches data from an S3 bucket, it verifies the ownership of the bucket using the bucket owner condition. By default, Data Prepper expects an S3 bucket to be owned by the same that owns the correlating SQS queue. When no SQS is provided, Data Prepper uses the Amazon Resource Name (ARN) role in the aws configuration.

If you plan to ingest data from multiple S3 buckets but each bucket is associated with a different S3 account, you need to configure Data Prepper to check for cross-account S3 access, according to the following conditions:

• If all S3 buckets you want data from belong to an account other than that of the SQS queue, set default_bucket_owner to the account ID of the bucket account holder.
• If your S3 buckets are in multiple accounts, use a bucket_owners map.

In the following example, the SQS queue is owned by account 000000000000. The SQS queue contains data from two S3 buckets: my-bucket-01 and my-bucket-02. Because my-bucket-01 is owned by 123456789012 and my-bucket-02 is owned by 999999999999, the bucket_owners map calls both bucket owners with their account IDs, as shown in the following configuration:

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

You can use both bucket_owners and default_bucket_owner together.

Configuration

You can use the following options to configure the s3 source.

sqs

The following parameters allow you to configure usage for Amazon SQS in the s3 source plugin.

aws

codec

The codec determines how the s3 source parses each S3 object.

newline codec

The newline codec parses each single line as a single log event. This is ideal for most application logs because each event parses per single line. It can also be suitable for S3 objects that have individual JSON objects on each line, which matches well when used with the parse_json processor to parse each line.

Use the following options to configure the newline codec.

json codec

The json codec parses each S3 object as a single JSON object from a JSON array and then creates a Data Prepper log event for each object in the array.

csv codec

The csv codec parses objects in comma-separated value (CSV) format, with each row producing a Data Prepper log event. Use the following options to configure the csv codec.

Using s3_select with the s3 source

When configuring s3_select to parse S3 objects, use the following options.

csv

Use the following options in conjunction with the csv configuration for s3_select to determine how your parsed CSV file should be formatted.

These options map directly to options available in the S3 Select CSVInput data type.

json

Use the following option in conjunction with json for s3_select to determine how S3 Select processes the JSON file.

Using scan with the s3 source

The following parameters allow you to scan S3 objects. All options can be configured at the bucket level.

bucket

You can configure the following options inside the bucket setting.

filter

Use the following options inside the filter configuration.

scheduling

Metrics

The s3 source includes the following metrics.

Counters

• s3ObjectsFailed: The number of S3 objects that the s3 source failed to read.
• s3ObjectsNotFound: The number of S3 objects that the s3 source failed to read due to an S3 “Not Found” error. These are also counted toward s3ObjectsFailed.
• s3ObjectsAccessDenied: The number of S3 objects that the s3 source failed to read due to an “Access Denied” or “Forbidden” error. These are also counted toward s3ObjectsFailed.
• s3ObjectsSucceeded: The number of S3 objects that the s3 source successfully read.
• sqsMessagesReceived: The number of Amazon SQS messages received from the queue by the s3 source.
• sqsMessagesDeleted: The number of Amazon SQS messages deleted from the queue by the s3 source.
• sqsMessagesFailed: The number of Amazon SQS messages that the s3 source failed to parse.
• s3ObjectNoRecordsFound – The number of S3 objects that resulted in 0 records added to the buffer by the s3 source.
• sqsMessagesDeleteFailed – The number of SQS messages that the s3 source failed to delete from the SQS queue.
• s3ObjectsDeleted – The number of S3 objects deleted by the s3 source.
• s3ObjectsDeleteFailed – The number of S3 objects that the s3 source failed to delete.

Timers

• s3ObjectReadTimeElapsed: Measures the amount of time the s3 source takes to perform a request to GET an S3 object, parse it, and write events to the buffer.
• sqsMessageDelay: Measures the time elapsed from when S3 creates an object to when it is fully parsed.

Distribution summaries

• s3ObjectSizeBytes: Measures the size of S3 objects as reported by the S3 Content-Length. For compressed objects, this is the compressed size.
• s3ObjectProcessedBytes: Measures the bytes processed by the s3 source for a given object. For compressed objects, this is the uncompressed size.
• s3ObjectsEvents: Measures the number of events (sometimes called records) produced by an S3 object.

Example: Uncompressed logs with sqs

The following pipeline.yaml file shows the minimum configuration for reading uncompressed newline-delimited logs:

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"

Example: Uncompressed logs with scan

The following pipeline.yaml file shows the minimum configuration for scanning objects with uncompressed newline-delimited logs:

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"