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>

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	`10`	Polling interval in seconds between SQS queue checks (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 archive (0 = unlimited)
`max_size_archive_bytes`	N	`104857600`	Maximum total size of archive contents in bytes (0 = unlimited, default: 100 MB)

SQS Configuration

Field	Required	Default	Description
`sqs_max_messages`	N	`1`	Maximum messages to receive per poll (1-10)
`sqs_visibility_timeout`	N	`30`	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.

Ingest Buffer

Field	Required	Default	Description
`max_buffer_size`	N	`128MB`	Ingest buffer admission cap. Also sets this listener's eager heap reservation

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, which commits 384 MiB of heap per listener worker before any data arrives — 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. Ignored when `cron` is set
`cron`	N	-	Cron expression for the list cycle, taking precedence over `interval`
`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 (single JSON object), .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.

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

Synopsis​

Schema​

Configuration​

Device​

AWS Connection​

File Processing​

SQS Configuration​

Authentication Methods​

Ingest Buffer​

Details​

Ingestion Modes​

Poll Mode Properties​

The Cursor​

Date Tokens in prefix​

IAM Permissions​

Replay​

Examples​

Basic IAM Role Authentication​

Access Key Authentication​

Cross-Account Role Assumption​

High-Volume Processing​

Parquet File Processing​

Archive Processing with Limits​

Pipeline Integration​