AWS S3 input

Use the aws-s3 input to retrieve logs from S3 objects that are pointed to by S3 notification events read from an SQS queue or directly polling list of S3 objects in an S3 bucket. The use of SQS notification is preferred: polling lists of S3 objects is expensive in terms of performance and costs and should be preferably used only when no SQS notification can be attached to the S3 buckets. This input can, for example, be used to receive S3 access logs to monitor detailed records for the requests that are made to a bucket. This input also supports S3 notification from SNS to SQS.

SQS notification method is enabled setting queue_url configuration value. S3 bucket list polling method is enabled setting bucket_arn configuration value. Both values cannot be set at the same time, at least one of the values must be set.

When using the SQS notification method, this input depends on S3 notifications delivered to an SQS queue for s3:ObjectCreated:* events. You must create an SQS queue and configure S3 to publish events to the queue.

The S3 input manages SQS message visibility to prevent messages from being reprocessed while the S3 object is still being processed. If the processing takes longer than half of the visibility timeout, the timeout is reset to ensure the message doesn’t return to the queue before processing is complete.

If an error occurs during the processing of the S3 object, the processing will be stopped, and the SQS message will be returned to the queue for reprocessing.

This example reads s3:ObjectCreated notifications from SQS, and assumes that all the S3 objects have a Content-Type of application/json. It splits the Records array in the JSON into separate events.

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

When using the direct polling list of S3 objects in an S3 buckets, a number of workers that will process the S3 objects listed must be set through the number_of_workers config. Listing of the S3 bucket will be polled according the time interval defined by bucket_list_interval config. The default value is 120 sec.

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

The aws-s3 input can also poll third party S3-compatible services such as the Minio. Using non-AWS S3 compatible buckets requires the use of access_key_id and secret_access_key for authentication. To specify the S3 bucket name, use the non_aws_bucket_name config and the endpoint must be set to replace the default API endpoint. endpoint should be a full URI in the form of https(s)://<s3 endpoint> in the case of non_aws_bucket_name, that will be used as the API endpoint of the service. No endpoint is needed if using the native AWS S3 service hosted at amazonaws.com. Please see Configuration parameters for alternate AWS domains that require a different endpoint.

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

This aws-s3 input feature prevents the duplication of events in Elasticsearch by generating a custom document _id for each event, rather than relying on Elasticsearch to automatically generate one. Each document in an Elasticsearch index must have a unique _id, and Filebeat uses this property to avoid ingesting duplicate events.

The custom _id is based on several pieces of information from the S3 object: the Last-Modified timestamp, the bucket ARN, the object key, and the byte offset of the data in the event.

Duplicate prevention is particularly useful in scenarios where Filebeat needs to retry an operation. Filebeat guarantees at-least-once delivery, meaning it will retry any failed or incomplete operations. These retries may be triggered by issues with the host, {beatname_uc}, network connectivity, or services such as Elasticsearch, SQS, or S3.

There are some limitations to consider when using _id-based deduplication in Elasticsearch:

If you want to disable the _id-based deduplication, you can remove the document _id using the drop_fields processor in Filebeat.

filebeat.inputs:
  - type: aws-s3
    queue_url: https://queue.amazonaws.com/80398EXAMPLE/MyQueue
    processors:
      - drop_fields:
          fields:
            - '@metadata._id'
          ignore_missing: true

Alternatively, you can remove the _id field using an Elasticsearch Ingest Node pipeline.

{
  "processors": [
    {
      "remove": {
        "if": "ctx.input?.type == \"aws-s3\"",
        "field": "_id",
        "ignore_missing": true
      }
    }
  ]
}

The aws-s3 input supports the following configuration options plus the Common options described later.

The maximum duration of the AWS API call. If it exceeds the timeout, the AWS API call will be interrupted. The default AWS API timeout is 120s.

The API timeout must be longer than the sqs.wait_time value.

The size of the buffer in bytes that each harvester uses when fetching a file. This only applies to non-JSON logs. The default is 16 KiB.

A standard MIME type describing the format of the object data. This can be set to override the MIME type given to the object when it was uploaded. For example: application/json.

The file encoding to use for reading data that contains international characters. This only applies to non-JSON logs. See encoding.

The file decoding option is used to specify a codec that will be used to decode the file contents. This can apply to any file stream data. An example config is shown below:

Currently supported codecs are given below:-

The CSV codec is used to decode RFC 4180 CSV data streams. Enabling the codec without other options will use the default codec options.

  decoding.codec.csv.enabled: true

The csv codec supports five sub attributes to control aspects of CSV decoding. The comma attribute specifies the field separator character used by the CSV format. If it is not specified, the comma character , is used. The comment attribute specifies the character that should be interpreted as a comment mark. If it is specified, lines starting with the character will be ignored. Both comma and comment must be single characters. The lazy_quotes attribute controls how quoting in fields is handled. If lazy_quotes is true, a quote may appear in an unquoted field and a non-doubled quote may appear in a quoted field. The trim_leading_space attribute specifies that leading white space should be ignored, even if the comma character is white space. For complete details of the preceding configuration attribute behaviors, see the CSV decoder documentation The fields_names attribute can be used to specify the column names for the data. If it is absent, the field names are obtained from the first non-comment line of data. The number of fields must match the number of field names.

An example config is shown below:

  decoding.codec.csv.enabled: true
  decoding.codec.csv.comma: "\t"
  decoding.codec.csv.comment: "#"

The parquet codec is used to decode the Apache Parquet data storage format. Enabling the codec without other options will use the default codec options.

  decoding.codec.parquet.enabled: true

The Parquet codec supports two attributes, batch_size and process_parallel, to improve decoding performance:

By adjusting both batch_size and process_parallel, you can fine-tune the trade-off between processing speed and memory consumption.

An example config is shown below:

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

If the fileset using this input expects to receive multiple messages bundled under a specific field or an array of objects then the config option expand_event_list_from_field value can be assigned the name of the field or .[]. This setting will be able to split the messages under the group value into separate events. For example, CloudTrail logs are in JSON format and events are found under the JSON object "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",
        }
    ]
}

Or when expand_event_list_from_field is set to .[], an array of objects will be split into separate events.

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

Note: When expand_event_list_from_field parameter is given in the config, aws-s3 input will assume the logs are in JSON format and decode them as JSON. Content type will not be checked. If a file has "application/json" content-type, expand_event_list_from_field becomes required to read the JSON file.

If the SQS queue will have events that correspond to files that Filebeat shouldn’t process file_selectors can be used to limit the files that are downloaded. This is a list of selectors which are made up of regex and expand_event_list_from_field options. The regex should match the S3 object key in the SQS message, and the optional expand_event_list_from_field is the same as the global setting. If file_selectors is given, then any global expand_event_list_from_field value is ignored in favor of the ones specified in the file_selectors. Regex syntax is the same as the Go language. Files that don’t match one of the regexes won’t be processed. content_type, parsers, include_s3_metadata,max_bytes, buffer_size, and encoding may also be set for each file selector.

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

Moved to AWS credentials options.

This input can include S3 object metadata in the generated events for use in follow-on processing. You must specify the list of keys to include. By default, none are included. If the key exists in the S3 response, then it will be included in the event as aws.s3.metadata.<key> where the key name as been normalized to all lowercase.

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

The maximum number of bytes that a single log message can have. All bytes after max_bytes are discarded and not sent. This setting is especially useful for multiline log messages, which can get large. This only applies to non-JSON logs. The default is 10 MiB.

This option expects a list of parsers that non-JSON logs go through.

Available parsers:

In this example, Filebeat is reading multiline messages that consist of XML that start with the <Event> tag.

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

See the available parser settings in detail below.

Options that control how Filebeat deals with log messages that span multiple lines. See Multiline messages for more information about configuring multiline options.

URL of the AWS SQS queue that messages will be received from. (Required when bucket_arn, access_point_arn, and non_aws_bucket_name are not set).

The name of the AWS region of the end point. If this option is given it takes precedence over the region name obtained from the queue_url value.

The duration that the received SQS messages are hidden from retrieve requests after being retrieved by a ReceiveMessage request. The default visibility timeout is 300s. The maximum is 12h. Filebeat will automatically reset the visibility timeout of a message after 1/2 of the duration passes to prevent a message that is still being processed from returning to the queue.

The maximum number of times a SQS message should be received (retried) before deleting it. This feature prevents poison-pill messages (messages that can be received but can’t be processed) from consuming resources. The number of times a message has been received is tracked using the ApproximateReceiveCount SQS attribute. The default value is 5.

If you have configured a dead letter queue, then you can set this value to -1 to disable deletion on failure.

Inline Javascript source code.

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;
  }

Path to a script file to load. Relative paths are interpreted as relative to the path.config directory. Globs are expanded.

This loads filter.js from disk.

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

List of script files to load. The scripts are concatenated together. Relative paths are interpreted as relative to the path.config directory. And globs are expanded.

A dictionary of parameters that are passed to the register of the script.

Parameters can be passed to the script by adding params to the config. This allows for a script to be made reusable. When using params the code must define a register(params) function to receive the parameters.

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;
    }

This sets an execution timeout for the process function. When the process function takes longer than the timeout period the function is interrupted. You can set this option to prevent a script from running for too long (like preventing an infinite while loop). By default, there is no timeout.

This sets the maximum number of JavaScript VM sessions that will be cached to avoid reallocation.

The maximum duration that an SQS ReceiveMessage call should wait for a message to arrive in the queue before returning. The default value is 20s. The maximum value is 20s.

ARN of the AWS S3 bucket that will be polled for list operation. (Required when queue_url, access_point_arn, and `non_aws_bucket_name are not set).

ARN of the AWS S3 Access Point that will be polled for list operation. (Required when queue_url, bucket_arn, and non_aws_bucket_name are not set).

Name of the S3 bucket that will be polled for list operation. Required for third-party S3 compatible services. (Required when queue_url and bucket_arn are not set).

Time interval for polling listing of the S3 bucket: default to 120s.

Prefix to apply for the list request to the S3 bucket. Default empty.

Number of workers that will process the S3 or SQS objects listed. Required when bucket_arn or access_point_arn is set, otherwise (in the SQS case) defaults to 5.

Name of the third-party S3 bucket provider like backblaze or GCP. The following endpoints/providers will be detected automatically:

Enabling this option sets the bucket name as a path in the API call instead of a subdomain. When enabled https://<bucket-name>.s3.<region>.<provider>.com becomes https://s3.<region>.<provider>.com/<bucket-name>;. This is only supported with third-party S3 providers. AWS does not support path style.

To make AWS API calls, aws-s3 input requires AWS credentials. Please see AWS credentials options for more details.

The ARN of the S3 bucket where processed files are copied. The copy is created after the S3 object is fully processed. When using the non_aws_bucket_name, please use non_aws_backup_to_bucket_name accordingly.

Naming of the backed up files can be controlled with backup_to_bucket_prefix.

This prefix will be prepended to the object key when backing it up to another (or the same) bucket.

The name of the non-AWS bucket where processed files are copied. Use this parameter when not using AWS buckets. The copy is created after the S3 object is fully processed. When using the bucket_arn, please use backup_to_bucket_arn accordingly.

Naming of the backed up files can be controlled with backup_to_bucket_prefix.

Controls whether fully processed files will be deleted from the bucket.

This option can only be used together with the backup functionality.

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

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

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

s3:GetObject
s3:ListBucket
s3:GetBucketLocation

s3:PutObject

s3:DeleteObject

sqs:GetQueueAttributes

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";
    }
}

<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"
    }
  }
}

$ 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