Skip to main content

aws_cloudwatch_logs

Consumes log events from AWS CloudWatch Logs.

# Common config fields, showing default values
input:
  label: ""
  aws_cloudwatch_logs:
    log_group_name: "" # No default (required)
    log_stream_names: [] # No default (optional)
    log_stream_prefix: "" # No default (optional)
    filter_pattern: "" # No default (optional)
    start_time: "" # No default (optional)
    poll_interval: "5s"
    auto_replay_nacks: true

Polls CloudWatch Log Groups for log events. Supports filtering by log streams, CloudWatch filter patterns, and configurable start times.

Each log event becomes a separate message with metadata including the log group name, log stream name, timestamp, and ingestion time.

IMPORTANT: This input tracks its position in memory only. If the process restarts, it will resume from the configured start_time (or the beginning if not set). For exactly-once processing, you should configure an appropriate start_time or implement idempotent downstream processing.

Credentials​

By default Expanso Edge will use a shared credentials file when connecting to AWS services. It's also possible to set them explicitly at the component level, allowing you to transfer data across accounts. You can find out more in this document.

Metadata​

This input adds the following metadata fields to each message:

  • cloudwatch_log_group - The name of the log group
  • cloudwatch_log_stream - The name of the log stream
  • cloudwatch_timestamp - The timestamp of the log event (Unix milliseconds)
  • cloudwatch_ingestion_time - The ingestion timestamp (Unix milliseconds)
  • cloudwatch_event_id - The unique event ID

You can access these metadata fields using Bloblang.

Fields​

log_group_name​

The name of the CloudWatch Log Group to consume from.

Type: string

log_stream_names​

An optional list of log stream names to consume from. If not set, events from all streams in the log group will be consumed.

Type: array of string

log_stream_prefix​

An optional log stream name prefix to filter streams. Only streams starting with this prefix will be consumed.

Type: string

filter_pattern​

An optional CloudWatch Logs filter pattern to apply when querying log events. See AWS documentation for filter pattern syntax.

Type: string

start_time​

The time to start consuming log events from. Can be an RFC3339 timestamp (e.g., 2024-01-01T00:00:00Z) or the string now to start consuming from the current time. If not set, starts from the beginning of available logs.

Type: string

poll_interval​

The interval at which to poll for new log events.

Type: string
Default: "5s"

limit​

The maximum number of log events to return in a single API call. Valid range: 1-10000.

Type: int
Default: 1000

structured_log​

Whether to output log events as structured JSON objects with all metadata fields, or as plain text messages with metadata in message metadata.

Type: bool
Default: true

api_timeout​

The maximum time to wait for an API request to complete.

Type: string
Default: "30s"

auto_replay_nacks​

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

Type: bool
Default: true

region​

The AWS region to target.

Type: string

endpoint​

Allows you to specify a custom endpoint for the AWS API.

Type: string

tcp​

TCP socket configuration.

Type: object

tcp.connect_timeout​

Maximum amount of time a dial will wait for a connect to complete. Zero disables.

Type: string
Default: "0s"

tcp.keep_alive​

TCP keep-alive probe configuration.

Type: object

tcp.keep_alive.idle​

Duration the connection must be idle before sending the first keep-alive probe. Zero defaults to 15s. Negative values disable keep-alive probes.

Type: string
Default: "15s"

tcp.keep_alive.interval​

Duration between keep-alive probes. Zero defaults to 15s.

Type: string
Default: "15s"

tcp.keep_alive.count​

Maximum unanswered keep-alive probes before dropping the connection. Zero defaults to 9.

Type: int
Default: 9

tcp.tcp_user_timeout​

Maximum time to wait for acknowledgment of transmitted data before killing the connection. Linux-only (kernel 2.6.37+), ignored on other platforms. When enabled, keep_alive.idle must be greater than this value per RFC 5482. Zero disables.

Type: string
Default: "0s"

credentials​

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Type: object

credentials.profile​

A profile from ~/.aws/credentials to use.

Type: string

credentials.id​

The ID of credentials to use.

Type: string

credentials.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

credentials.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

credentials.from_ec2_role​

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

credentials.role​

A role ARN to assume.

Type: string

credentials.role_external_id​

An external ID to provide when assuming a role.

Type: string

Delivery guarantee​

This input is at-least-once. Its read position is held in memory only: while running it advances a cursor past the highest log-event ingestion time it has seen and keeps the pagination token for the page it is working through, and neither is written to disk.

On restart the input begins again from the configured start_time, or from the log group's earliest retained events when start_time is unset. Everything between that point and wherever the previous run had reached is delivered a second time.

start_time is not a checkpoint

start_time selects where a fresh run begins. It does not record where the last run stopped, so it cannot bound duplicates after a restart: an earlier value replays more, a later one silently skips events that arrived before it. Plan for repeats rather than trying to tune them away.

Make the downstream side idempotent, or deduplicate on arrival. Every message carries cloudwatch_event_id, which is unique per log event and stable across redelivery, so it is the key to use:

pipeline:
  processors:
    - dedupe:
        cache: seen_events
        key: '${! @cloudwatch_event_id }'

cache_resources:
  - label: seen_events
    memory:
      default_ttl: 6h

Two separate limits decide whether that cache actually stops duplicates:

  • A memory cache is per process and resets with it, so it collapses duplicates within a run but not across a restart. Deduplicating across restarts needs a shared cache.
  • default_ttl has to outlive the replay window. A restart re-reads from start_time, so the cache still has to hold the id of the oldest event that can come back. Any entry that expired in the meantime lets its event through as new. The 6h above is only correct if nothing older than six hours can be replayed.

How wide that window is depends on start_time:

start_timeReplay window after a restart
unset (the default)the beginning of available logs, so the log group's whole retention period
an RFC3339 timestampeverything since that instant, so the window grows for as long as the pipeline runs
nownothing from before the restart, at the cost of losing whatever arrived while the pipeline was down

For the first two there is no TTL worth paying for: covering them means holding every event id for the group's retention. Enforce uniqueness in the destination instead, through a unique constraint or a write keyed by cloudwatch_event_id, and treat the cache as a same-run optimisation rather than the thing standing between you and duplicates.