Amazon S3

Amazon AWS Long-Term Storage

Synopsis

Amazon S3 device processes files from Amazon S3 buckets using SQS event notifications. Director polls an SQS queue at configured intervals to receive S3 object-created events, downloads the referenced objects, and processes them through DataStream pipelines. The device supports multiple file formats including JSON, JSONL, Parquet, and compressed archives.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: awss3
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    region: <string>
    endpoint: <string>
    use_path_style: <boolean>
    access_key_id: <string>
    secret_access_key: <string>
    session_token: <string>
    queue_url: <string>
    role_arn: <string>
    bucket: <string>
    object_size: <numeric>
    timeout: <numeric>
    file_name_filter: <string>
    max_files_in_archive: <numeric>
    max_size_archive_bytes: <numeric>
    sqs_max_messages: <numeric>
    sqs_visibility_timeout: <numeric>
    sqs_wait_time_seconds: <numeric>
    mode: <enum>
    prefix: <string>
    workers: <numeric>
    event_split_rules: <string[]>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description of the device's purpose
type	Y	-	Device type identifier (must be awss3)
tags	N	-	Array of labels for categorization
pipelines	N	-	Array of preprocessing pipeline references
status	N	true	Boolean flag to enable/disable the device

AWS Connection

Field	Required	Default	Description
region	Y	-	AWS region where S3 bucket and SQS queue are located
endpoint	N	-	Custom endpoint URL for the S3, SQS, and STS clients (e.g., LocalStack). Leave empty for standard AWS
use_path_style	N	computed	Use path-style S3 addressing. Derived from endpoint when unset: false for an empty endpoint or an amazonaws.com host, true for any other endpoint — so a custom S3-compatible endpoint (MinIO, LocalStack, Ceph) gets path-style without configuration. Set it explicitly to override
queue_url	Y	-	SQS queue URL that receives S3 event notifications
access_key_id	Y*	-	AWS access key ID for static-credential authentication
secret_access_key	Y*	-	AWS secret access key for static-credential authentication
session_token	N	-	AWS session token for temporary credentials
role_arn	N	-	AWS IAM role ARN to assume for cross-account access
bucket	N	-	S3 bucket name. Used only for startup validation (HeadBucket when set, ListBuckets when empty); does not restrict event processing
object_size	N	100000000	Maximum S3 object size in bytes to download (default: 100 MB)

* = Conditionally required (see Authentication Methods below). access_key_id and secret_access_key must be provided together when using static credentials.

File Processing

Field	Required	Default	Description
timeout	N	1	queue mode only. Idle backoff in seconds when the queue is empty and long polling is off (sqs_wait_time_seconds: 0); with long polling on, SQS does the waiting and this value is not used (1-10)
file_name_filter	N	".*"	Regular expression filter for S3 object keys
mode	N	queue	Ingestion mode: queue or poll. See Ingestion Modes
prefix	N	-	Restrict ingestion to keys under this prefix. In poll mode it may carry date tokens
workers	N	max(NumCPU/4, 1)	Concurrent download and decode workers, 1-64
max_files_in_archive	N	100	Maximum number of files to process from an archive. 0 or an empty value uses the default
max_size_archive_bytes	N	104857600	Maximum total size of archive contents in bytes (100 MB). 0 or an empty value uses the default

SQS Configuration

Field	Required	Default	Description
sqs_max_messages	N	10	Maximum messages to receive per poll (1-10)
sqs_visibility_timeout	N	300	Message visibility timeout in seconds (0-43200)
sqs_wait_time_seconds	N	20	Long polling wait time in seconds (0-20)

Authentication Methods

The device supports multiple authentication methods:

• IAM role: Omit credentials to use the instance/task IAM role.
• Access keys: Provide access_key_id and secret_access_key for static credentials.
• Role Assumption: Provide role_arn to assume a different IAM role.
• Temporary Credentials: Include session_token alongside access keys for temporary security credentials.

Credential fields (access_key_id, secret_access_key, session_token, role_arn) support ${ENV_VAR} and $secret{...} token resolution.

Event Split Rules

Field	Required	Default	Description
event_split_rules	N	-	Ordered list of rule set keys that split each file into separate logs. See Event Split Rules

In the console, the list is Rule sets under Event split rules, at the end of the SQS settings step and on the SQS settings tab. A new device starts with none, and then reads each file by its file extension.

The unit is the file, not the message. The Director checks the rules once for each file, against the start of the file after it is decompressed: as many bytes as the rule set's Bytes to inspect (inspectBytes) allows. The first rule that matches splits the whole file as the Director reads it, so a large file is never held in memory whole. A condition reads the file's path, <bucket>/<key>, as source. A file that no rule matches is read as before: by the decoder for its extension, and the device's split_mode then splits each record. Each file inside an archive is matched on its own, and Parquet files are never matched. A file of an extension the device does not read, such as .csv, is refused before the rules are checked, so a matching rule does not make it readable. The device still acknowledges a file as a whole: its queue message, or its place in the listing, waits until all of the file's logs are delivered.

A Director reads the list from version 2.5.1. An older one ignores it and reads each file by its extension. The awss3replay device does not take the setting. Changing the list restarts the device; editing a rule set it uses does not.

Ingest Buffer

Field	Required	Default	Description
max_buffer_size	N	128MB	Ingest buffer admission cap. Also sizes this listener's ingest buffers: 3/8 of this value at start, growing to at most 3x

note

max_buffer_size is the ingest-buffer admission cap. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

Details

Ingestion Modes

This device offers two ingestion modes, selected with mode:

mode	Behaviour
queue	Default. The device consumes notifications from the bucket's event queue and fetches each object as it is announced.
poll	The device lists the bucket on a schedule and ingests what it has not seen before. No queue is needed.

Any other value fails with mode field not valid.

Poll mode suits a bucket with no notification wiring, and a historical backfill, since a listing can be bounded to a date range. It costs a list operation per cycle, so it is the wrong choice for a high-rate bucket that already has a queue.

Poll Mode Properties

These are read only when mode: poll is set.

Field	Required	Default	Description
interval	N	60	Seconds between list cycles. When cron is set as well, a cycle starts when either is due. The default applies only when cron is unset
cron	N	-	Cron expression for the list cycle. Alone, it is the only schedule; with interval, a cycle starts when either is due. An invalid expression is logged and the interval schedule is used instead
workers	N	max(NumCPU/4, 1)	Concurrent download and decode workers, 1–64
max_retries	N	0	Consecutive cycles an object may fail transiently before the poller gives up on it and moves the cursor past it. 0 means never give up
start_date	N	-	Lower bound of the listing window. Accepts Unix seconds, RFC 3339, a timestamp without an offset (read in timezone), or a relative expression. Unset or negative means the whole bucket
end_date	N	-	Upper bound, exclusive. Unset means unbounded; a negative value is an error, and a value at or before start_date is rejected
timezone	N	UTC	Zone used to interpret an offset-less start_date or end_date

The Cursor

Poll mode keeps a since-cursor so a restart or a cluster failover does not re-ingest what it already has. The cursor is scoped to the listing target — the bucket and prefix together — so changing either starts a fresh one.

A stored cursor takes precedence over start_date, but only while the window is unchanged: editing start_date, end_date or prefix discards the cursor and re-evaluates the new window. A bounded window that has been fully drained stops listing and idles until the window changes.

Date Tokens in prefix

In poll mode prefix may carry date tokens, so a listing walks only the partitions that fall inside its window: %Y, %y, %m, %d, %j, %H, and %% for a literal percent. The finest token present sets the partition step.

A token-bearing prefix needs a bounded start_date — there is no way to walk partitions from the beginning of a bucket — and a window spanning too many partitions is rejected rather than expanded.

warning

Date tokens are rejected in queue mode. A queue subscription filters on a fixed key prefix, which cannot be re-expanded per cycle, so a prefix containing %Y and the like fails at apply time. Write %% where a literal percent is wanted.

IAM Permissions

When using IAM role-based authentication, the following permissions are required:

IAM Action	Purpose
s3:GetObject	Download S3 objects for processing
s3:ListBucket	Validate bucket access at startup (when bucket name is configured)
s3:ListAllMyBuckets	Validate S3 access at startup (when bucket name is not configured)
sqs:ReceiveMessage	Poll SQS queue for S3 event notifications
sqs:DeleteMessage	Remove processed messages from the queue
sqs:GetQueueAttributes	Validate SQS queue connectivity at startup

When using cross-account role assumption (role_arn), the calling identity also requires:

IAM Action	Purpose
sts:AssumeRole	Assume IAM role in the target account

Minimum IAM Policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3ReadObjects",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::BUCKET_NAME/*"
    },
    {
      "Sid": "S3ValidateBucket",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::BUCKET_NAME"
    },
    {
      "Sid": "SQSConsumeMessages",
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "arn:aws:sqs:REGION:ACCOUNT_ID:QUEUE_NAME"
    }
  ]
}

Cross-Account Access

When accessing S3 buckets in another AWS account, configure role_arn and optionally use temporary credentials. The assumed role must have the S3 and SQS permissions above. The target role's trust policy must allow assumption from the source account, with optional ExternalId condition for Security Lake scenarios.

The Amazon S3 device processes S3 objects based on SQS notifications. The device continuously polls an SQS queue for S3 event messages, downloads the referenced objects, and processes their contents through the telemetry pipeline.

Event Processing Flow: The device receives S3 event notifications from SQS containing bucket name and object key information. For each ObjectCreated event (Put, Post, Copy, CompleteMultipartUpload), the device downloads the S3 object and processes it according to its file type. After successful processing, the SQS message is deleted to prevent reprocessing.

File Format Detection: The device automatically detects file format based on file extension. Supported formats include .json (one JSON object, an array, or several JSON values such as one object per line), .jsonl (newline-delimited JSON), .parquet / .parq / .pq (columnar format), and compressed archives (.gz, .zip, .bz2, .tar, .tgz, .tar.gz, .tar.bz2). For .log and .txt files, the device performs content auto-detection by examining the first 8KB of data. A .gz file whose content is not compressed is read as the file its name stands for, so a plain PT1H.json.gz is read as PT1H.json.

JSON Files: A .json file can hold more than one JSON value. Azure writes its diagnostic logs to storage with one record per line (insights-logs-<category>/.../PT1H.json), and the File, Amazon S3, Azure Blob Storage and Google Cloud Storage targets write their default json format the same way. The device reads every value in the file:

File content	Records
One JSON object	One record
A JSON array	One record for each element
Several JSON values one after another, such as one object per line (JSON lines)	One record for each value
A mix, such as an object followed by an array	One record for each object, and one for each element of each array
A value partway through that is not valid JSON	The records before it are delivered. Reading stops there, and the device logs an error that names the file.

The same rules apply to a .json file inside an archive, and to a .log or .txt file of JSON lines whose first line is longer than 8KB: content detection reads only the first 8KB, so such a file is read as JSON.

A record keeps the text the file has. Keys stay in their order, large integers keep every digit, and characters such as < and & are not escaped. The white space between the parts of a record is removed, so each record is one line.

When reading stops at a value that is not valid JSON, or at a file that ends partway through a record, the device still counts the file as read and does not retry it. The complete records before that point are delivered and the ones after it are not. The error in the director's log names the file, so you can fix it and upload it again. A corrected copy of the whole file delivers the records before that point a second time. To avoid duplicates, upload a file that holds only the records that were not delivered.

note

Earlier versions kept only the first record of a .json file that held more than one JSON value, and dropped the rest without an error. If a bucket of JSON lines files showed one record per file, this is the cause.

Archive Processing: Compressed archives are automatically extracted and processed. The device supports nested archives and applies file name filtering to extracted contents. Archive processing is controlled by max_files_in_archive and max_size_archive_bytes limits to prevent resource exhaustion from maliciously large archives.

Error Handling: The device distinguishes between transient and permanent errors. Transient errors (network issues, throttling) leave the SQS message in the queue for automatic retry. Permanent errors (unsupported format, file name filter mismatch, archive size limits) delete the message to prevent infinite retry loops.

SQS Integration: The device uses SQS long polling to efficiently wait for new messages. The sqs_wait_time_seconds parameter enables long polling to reduce API calls and costs. Message visibility timeout ensures that failed processing attempts don't block other consumers.

Replay

The awss3replay device type is this collector permanently in poll mode. It lists the bucket for a bounded time window and ingests the objects it finds, without an SQS notification queue.

Every property behaves as documented under Poll Mode above, including start_date, end_date, timezone, the date tokens in prefix, and the cursor. The only configuration difference is type: mode may be omitted, since a replay can only poll, and setting it to queue is rejected.

What the replay type adds is a reported run — live progress, a run history, and the pause, resume, cancel and run-now controls. See Replay Devices.

Examples

Basic IAM Role Authentication

Configuring Amazon S3 device using IAM role for authentication (recommended for EC2/ECS deployments)...	- id: 1 name: s3-cloudtrail-logs type: awss3 properties: region: "us-east-1" queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/cloudtrail-events"
Device polls SQS queue for S3 events and processes CloudTrail log files using instance IAM role...	{ "eventVersion": "1.08", "userIdentity": { "type": "IAMUser", "userName": "admin" }, "eventTime": "2024-01-15T10:30:00Z", "eventSource": "s3.amazonaws.com", "source_file": "AWSLogs/123456789012/CloudTrail/us-east-1/2024/01/15/log.json" }

Access Key Authentication

Using static AWS access keys for authentication with file name filtering...	- id: 2 name: s3-application-logs type: awss3 properties: region: "us-west-2" access_key_id: "AKIAIOSFODNN7EXAMPLE" secret_access_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" queue_url: "https://sqs.us-west-2.amazonaws.com/123456789012/app-logs" file_name_filter: "prod/.*\\.jsonl$"
Device processes only JSONL files from the 'prod/' prefix, filtering out other files...	{ "timestamp": "2024-01-15T10:30:00Z", "level": "ERROR", "message": "Database connection timeout", "source_file": "prod/app-errors-2024-01-15.jsonl" }

Cross-Account Role Assumption

Assuming an IAM role in a different AWS account for cross-account S3 access...	- id: 3 name: s3-partner-data type: awss3 properties: region: "eu-west-1" role_arn: "arn:aws:iam::987654321098:role/PartnerDataAccess" queue_url: "https://sqs.eu-west-1.amazonaws.com/987654321098/partner-events" timeout: 5
Device assumes the specified role to access S3 buckets in the partner account...	{ "partner_id": "partner-123", "event_type": "transaction", "amount": 1500.00, "timestamp": "2024-01-15T10:30:00Z", "source_file": "partner-data/transactions-2024-01-15.json" }

High-Volume Processing

Optimizing for high-volume S3 data processing with batch message retrieval...	- id: 4 name: s3-high-volume type: awss3 properties: region: "us-east-1" queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/bulk-data" timeout: 1 sqs_max_messages: 10 sqs_visibility_timeout: 300 sqs_wait_time_seconds: 20
Device retrieves up to 10 messages per poll with 5-minute visibility timeout for efficient high-volume processing...	{ "record_id": "rec_001", "metric_value": 42.5, "timestamp": "2024-01-15T10:30:00Z", "processing_info": { "batch_size": 10, "poll_interval": 1 } }

Parquet File Processing

Processing Parquet files from S3 for analytics data ingestion...	- id: 5 name: s3-analytics-parquet type: awss3 properties: region: "us-east-1" queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/analytics-events" file_name_filter: ".*\\.parquet$"
Device processes only Parquet files using columnar format for efficient large-dataset handling...	{ "user_id": "user_12345", "page_views": 42, "session_duration": 1800, "timestamp": "2024-01-15T10:30:00Z", "source_file": "analytics/user-behavior-2024-01-15.parquet" }

Archive Processing with Limits

Processing compressed archives with safety limits to prevent resource exhaustion...	- id: 6 name: s3-compressed-logs type: awss3 properties: region: "us-east-1" queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/log-archives" file_name_filter: ".*\\.(gz|zip)$" max_files_in_archive: 50 max_size_archive_bytes: 52428800
Device extracts and processes compressed archives, limiting to 50 files and 50 MB total size...	{ "log_entry": "Application started successfully", "timestamp": "2024-01-15T10:30:00Z", "source_file": "logs/app-2024-01-15.log.gz", "archive_info": { "total_files": 12, "total_size_bytes": 8388608 } }

Pipeline Integration

Integrating S3 device with preprocessing pipelines for data transformation...	- id: 7 name: s3-with-pipeline type: awss3 tags: - "aws" - "s3" pipelines: - timestamp-normalization - field-enrichment properties: region: "us-east-1" queue_url: "https://sqs.us-east-1.amazonaws.com/123456789012/raw-data"
S3 data is processed through normalization and enrichment pipelines before routing to targets...	{ "timestamp": "2024-01-15T10:30:00.000Z", "event_type": "user_login", "user_id": "user_456", "enriched_data": { "normalized_timestamp": "2024-01-15T10:30:00Z", "event_category": "authentication", "source_bucket": "raw-data" } }