CloudWatch Logs

LocalEmu CloudWatch Logs implements all 107 operations. Log groups, streams, events, sequence tokens, retention metadata, and KMS key associations are served by the moto-ext metadata backend, which holds everything in-process. On top of that, 11 operations have LocalEmu-custom code: PutLogEvents evaluates metric filters and forwards matches to CloudWatch in real time; the subscription-filter pipeline gzip+base64-wraps matching events and pushes them to Lambda, Kinesis, or Firehose synchronously; DescribeLogGroups, DescribeLogStreams, and ListLogGroups are reimplemented for prefix/pattern semantics and pagination; and the tag operations are LocalEmu-side with a dedicated store.

Operation-level coverage: see the CloudWatch Logs coverage matrix.

Quick start

$ awsemu logs create-log-group --log-group-name app/api
$ awsemu logs create-log-stream --log-group-name app/api --log-stream-name prod

$ awsemu logs put-log-events --log-group-name app/api --log-stream-name prod \
    --log-events '[
      {"timestamp": '$(date +%s000)', "message": "INFO request_id=abc status=200"},
      {"timestamp": '$(date +%s000)', "message": "ERROR request_id=def status=500"}
    ]' --query 'nextSequenceToken' --output text
00000000000000000000000000000000000000000000000001

$ awsemu logs filter-log-events --log-group-name app/api \
    --filter-pattern 'ERROR' --query 'events[].message'
["ERROR request_id=def status=500"]

$ awsemu logs describe-log-groups \
    --query 'logGroups[].{Name:logGroupName,Class:logGroupClass,Bytes:storedBytes}'
[{"Name": "app/api", "Class": "STANDARD", "Bytes": 64}]

Events land in the moto backend immediately, in insertion order per stream. Sequence tokens are monotonic and returned in the response. storedBytes on the group is the sum of message lengths across its streams.

Architecture

Code lives at services/logs/. The provider class LogsProvider at provider.py:59 combines two integration styles:

• •
  3 @handler methods override moto on the request boundary (DescribeLogGroups at provider.py:107, DescribeLogStreams at 136, ListLogGroups at 156).
• •
  6 @patch decorators reach into the moto backend itself (PutSubscriptionFilter at provider.py:413, MotoLogStream.put_log_events at 488, filter_log_events at 583, create_log_stream at 599, to_describe_dict at 607, get_log_events at 627). The patch path is required because subscription dispatch and filter-pattern matching happen inside moto's internal write path.
• •
  All other operations (95+ ops, lifecycle, retention, KMS associations, destinations, queries, anomaly detectors, account policies) delegate to moto unmodified.
• •
  Custom state in LogsStore (models.py:11-13) is a single TAGS dict mapping resource ARN to tag dict, accessed via the global logs_stores AccountRegionBundle. Tags live LocalEmu-side because moto's tag handling does not match the AWS shape across both Log-Group and Destination resources.

Persistence: accept_state_visitor (provider.py:64-68) declares both logs_backends and logs_stores as state roots. With PERSISTENCE=1, log groups, streams, events, subscription filters, metric filters, retention settings, KMS associations, and tags survive restart. In-flight subscription-filter dispatches do not survive (the next matching PutLogEvents after restart resumes normally).

Features supported

Subscription filters: Lambda, Kinesis, Firehose

PutSubscriptionFilter (provider.py:413-485) validates the destination at create time. It dispatches by ARN service segment:

• •
  :lambda:: verifies GetFunction succeeds against the destination; roleArn, if supplied, must not be set for vendor Lambda destinations.
• •
  :kinesis:: verifies the stream exists and is ACTIVE.
• •
  :firehose:: verifies the delivery stream exists.
• •
  Anything else: rejected with InvalidParameterException.

Once attached, every PutLogEvents evaluates the filter pattern against each event (provider.py:497-511). Matching events are wrapped in the AWS-standard envelope, gzip-compressed, base64-encoded, and forwarded synchronously:

{
  "messageType": "DATA_MESSAGE",
  "owner":       "<account-id>",
  "logGroup":    "app/api",
  "logStream":   "prod",
  "subscriptionFilters": ["errors-only"],
  "logEvents":   [{"id": "...", "timestamp": ..., "message": "..."}, ...]
}

For Lambda, the envelope is wrapped one more level as {"awslogs": {"data": "<base64>"}} and delivered via Invoke. For Kinesis and Firehose, the raw gzip payload is sent as the record body (provider.py:553-578). If a roleArn was supplied on the filter, the destination client is built via connect_to.with_assumed_role.

# 1. Create a Lambda that consumes the awslogs envelope
$ awsemu lambda create-function --function-name log-shipper \
    --runtime python3.12 --handler index.handler \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --zip-file fileb://shipper.zip

# 2. Wire it as a subscription on the log group
$ awsemu logs put-subscription-filter \
    --log-group-name app/api --filter-name errors-only \
    --filter-pattern 'ERROR' \
    --destination-arn arn:aws:lambda:us-east-1:000000000000:function:log-shipper

# 3. Any matching event triggers the Lambda in real time.
#    The function receives the standard gzip+base64 envelope:
#    { "awslogs": { "data": "<base64(gzip(json))>" } }
$ awsemu logs put-log-events --log-group-name app/api --log-stream-name prod \
    --log-events '[{"timestamp": '$(date +%s000)', "message": "ERROR boom"}]' >/dev/null

$ awsemu logs filter-log-events \
    --log-group-name /aws/lambda/log-shipper --limit 1 --query 'events[0].message'
"Got: {\"messageType\":\"DATA_MESSAGE\",\"logGroup\":\"app/api\",\"logEvents\":[{...}]}"

Metric filters: cross-publish to CloudWatch

Every PutLogEvents call walks the group's metric filters (provider.py:70-105). For each event matching a filter pattern, every transformation in metricTransformations emits one PutMetricData call against cloudwatch.PutMetricData with the configured metricNamespace, metricName, and value. The value is parsed as a float when numeric; the special $size placeholder is recognised but emits a warning (it currently falls back to value 1). Metric filters are skipped silently when the cloudwatch service is disabled.

$ awsemu logs put-metric-filter --log-group-name app/api \
    --filter-name error-count \
    --filter-pattern 'ERROR' \
    --metric-transformations '[{
      "metricName":      "ApiErrorCount",
      "metricNamespace": "App/Api",
      "metricValue":     "1"
    }]'

$ awsemu logs put-log-events --log-group-name app/api --log-stream-name prod \
    --log-events '[
      {"timestamp": '$(date +%s000)', "message": "ERROR a"},
      {"timestamp": '$(date +%s000)', "message": "INFO  b"},
      {"timestamp": '$(date +%s000)', "message": "ERROR c"}
    ]' >/dev/null

# The two ERROR matches landed in CloudWatch as a metric data point.
$ awsemu cloudwatch get-metric-statistics --namespace App/Api \
    --metric-name ApiErrorCount --statistics Sum \
    --start-time $(date -u -v-5M +%FT%TZ) --end-time $(date -u +%FT%TZ) \
    --period 60 --query 'Datapoints[].Sum'
[2.0]

Filter pattern syntax

The matcher at provider.py:301-399 implements the term-based syntax from the AWS docs. The same matcher is shared between FilterLogEvents, subscription-filter dispatch, and metric-filter evaluation, so behaviour is consistent across all three.

• •
  Empty pattern: matches every event (AWS-compatible).
• •
  Space-separated terms: implicit AND. All required terms must appear in the message.
• •
  Quoted phrases: "foo bar" matches the substring exactly.
• •
  -term: exclude. Event must not contain the term.
• •
  ?term: at-least-one-of. At least one ?-prefixed term must match.
• •
  JSON selectors: { $.path = "value" } or { $.path = 42 }. Operators = and !=, value types: quoted string, integer, float, true/false/null. Path traversal walks nested objects via dot.
• •
  Fallback: unsupported JSON expressions degrade to substring on the raw message (the AWS service rejects them; LocalEmu's relaxed fallback is documented and consistent).

# Substring / AND across space-separated terms
$ awsemu logs filter-log-events --log-group-name app/api \
    --filter-pattern 'ERROR request_id' --query 'events[].message'
["ERROR request_id=def status=500"]

# Exclude with -term  +  at-least-one-of with ?term
$ awsemu logs filter-log-events --log-group-name app/api \
    --filter-pattern '?ERROR ?WARN -healthcheck'

# JSON selector: only events whose JSON message has status >= 500
$ awsemu logs filter-log-events --log-group-name app/api \
    --filter-pattern '{ $.status = 500 }' --query 'events[].message'
["{\"status\":500,\"path\":\"/users\"}"]

Lambda invocation logs

Every Lambda invocation streams its stdout and stderr into a CloudWatch log group named /aws/lambda/<function-name>. The plumbing lives at services/lambda_/invocation/logs.py:

• •
  LogHandler at logs.py:29 runs a queue-driven worker thread per function.
• •
  The thread calls PutLogEvents at logs.py:61; on ResourceNotFoundException it creates the group + stream on the fly and retries (logs.py:66-79).
• •
  The logs client is built via connect_to.with_assumed_role using the function's execution role.
• •
  When the logs service is disabled, log shipping is a no-op (logs.py:89-91).

$ awsemu lambda create-function --function-name greet \
    --runtime python3.12 --handler index.handler \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --zip-file fileb://greet.zip

$ awsemu lambda invoke --function-name greet --payload '{"name":"Tarek"}' /tmp/out.json

# Stdout + stderr from the function were streamed into the log group
# /aws/lambda/<function-name> with one stream per invocation cohort.
$ awsemu logs filter-log-events \
    --log-group-name /aws/lambda/greet --limit 3 \
    --query 'events[].message'
[
  "START RequestId: 1234... Version: $LATEST",
  "Hello, Tarek",
  "END RequestId: 1234..."
]

Because the Lambda log path is just a CloudWatch Logs client, subscription filters and metric filters attached to /aws/lambda/<function-name> work the same way as on any user group.

VPC flow logs

When VPC flow logging is enabled on the Docker-backed network stack, the recorder at services/ec2/docker/flow_log_recorder.py buffers per-flow lines in-process and flushes every 60 seconds to the log group /localemu/vpc-flow-logs (stream name flow-YYYY-MM-DD). The group and stream are created on first flush; events use the AWS flow-log line format. From there, all normal CloudWatch Logs features apply: filter, subscribe to a Lambda, or wire a metric filter against the line shape.

Configuration

CloudWatch Logs has no service-specific environment variables in LocalEmu. Disable it globally with SERVICES=...,-logs (which also disables Lambda log shipping and VPC flow log forwarding). Disable just the cross-publish to CloudWatch metrics by disabling the cloudwatch service: PutLogEvents still runs, but the metric-filter loop is skipped (provider.py:81).

Integration points

Other LocalEmu services produce or consume from CloudWatch Logs.

Known limitations

• •
  Retention is metadata only. PutRetentionPolicy stores retentionInDays and DescribeLogGroups returns it, but no background worker evicts events past the window. Events live until process exit (or are restored on the next start with PERSISTENCE=1).
• •
  Live Tail (StartLiveTail) is a stub. The operation returns a successful session, but the WebSocket-style event stream does not push events in real time. Use FilterLogEvents in a poll loop for equivalent behaviour.
• •
  Logs Insights queries (StartQuery/GetQueryResults) are stubbed. The query lifecycle is tracked, but the query language is not parsed: results come back empty. The metadata operations (PutQueryDefinition, DescribeQueryDefinitions, DescribeQueries) work.
• •
  Data protection (PII masking) is metadata only. PutDataProtectionPolicy stores the policy; GetDataProtectionPolicy returns it; but log event payloads are not actually masked at read time.
• •
  KMS encryption is metadata only. AssociateKmsKey stores the kmsKeyId and DescribeLogGroups returns it, but events are not encrypted at rest under that key.
• •
  The $size metric value placeholder falls back to 1. A metric transformation with metricValue containing $size emits an info-level log and publishes value 1 instead of the matched-byte size.
• •
  Anomaly detectors are stubbed. CreateLogAnomalyDetector, ListLogAnomalyDetectors, and ListAnomalies succeed at the metadata level; no anomaly detection actually runs against ingested events.