AWS CloudWatch Log Exporter

AWS CloudWatch Log Exporter

Problem Statement

When debugging Lambda functions, ECS tasks, or any AWS service that writes to CloudWatch Logs, developers and DevOps engineers constantly find themselves clicking through the AWS Console or running ad-hoc aws logs filter-log-events commands. The console is slow, paginates awkwardly, and makes it painful to grab a chunk of logs for local analysis, share with a teammate, or feed into a grep pipeline. Meanwhile, the AWS CLI requires you to know exact timestamps, log stream names, and millisecond epoch values.

This script solves that by providing a single, ergonomic command that pulls logs from a CloudWatch log group for a relative time window (e.g. "last 30 minutes"), automatically discovers all streams in that window, handles pagination transparently, and writes a clean, sortable output to stdout or a file. It's the tool you want when someone says "just send me the last hour of logs from that Lambda."

How It Works

The script uses boto3's logs client and performs a three-phase fetch. First, it computes a start/end epoch-millisecond window from a human-friendly --since argument like 1h, 30m, or 2d. Second, it calls describe_log_streams with orderBy=LastEventTime and descending=True to enumerate the streams that had activity in that window. Third, for each stream it calls filter_log_events (which handles pagination via nextToken) and collects events into a list. Finally it sorts everything chronologically and emits either human-readable lines or JSON.

Key design decisions: (1) we use filter_log_events rather than get_log_events because it works at the log-group level and supports time filtering natively; (2) we cap the number of streams and events to avoid runaway API calls on huge groups; (3) we use ThreadPoolExecutor to fetch streams in parallel, which dramatically speeds up multi-stream groups; (4) all timestamps are emitted in ISO-8601 UTC so downstream tools can parse them. The script follows a functional core / imperative shell pattern — the fetch logic is pure-ish and testable, the CLI is thin.

Full Script

#!/usr/bin/env python3
"""
AWS CloudWatch Log Exporter

Fetch logs from an AWS CloudWatch log group for a relative time window
and print them to stdout (or a file) as text or JSON.

Usage:
    python3 cw_logs.py --group /aws/lambda/myfunction --since 1h
    python3 cw_logs.py --group /ecs/my-service --since 30m --format json
    python3 cw_logs.py --group /aws/lambda/myfunction --since 2h --grep ERROR

Requires: boto3, valid AWS credentials in the environment.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable

import boto3
from botocore.config import Config as BotoConfig
from botocore.exceptions import BotoCoreError, ClientError

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
AWS_REGION = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
MAX_STREAMS = int(os.environ.get("CW_MAX_STREAMS", "50"))
MAX_EVENTS = int(os.environ.get("CW_MAX_EVENTS", "10000"))
MAX_WORKERS = int(os.environ.get("CW_MAX_WORKERS", "8"))
BOTO_RETRIES = int(os.environ.get("CW_BOTO_RETRIES", "5"))

# Pattern for --since arguments: 30s, 15m, 2h, 7d
_SINCE_RE = re.compile(r"^(\d+)([smhd])$")
_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400}


def parse_since(value: str) -> timedelta:
    """Parse a relative duration like '30m', '2h', '7d' into a timedelta."""
    match = _SINCE_RE.match(value.strip().lower())
    if not match:
        raise argparse.ArgumentTypeError(
            f"Invalid --since value {value!r}. Use forms like 30s, 15m, 2h, 7d."
        )
    amount, unit = match.groups()
    return timedelta(seconds=int(amount) * _UNIT_SECONDS[unit])


def build_client(region: str):
    """Create a boto3 CloudWatch Logs client with adaptive retries."""
    cfg = BotoConfig(retries={"max_attempts": BOTO_RETRIES, "mode": "adaptive"})
    return boto3.client("logs", region_name=region, config=cfg)


def list_active_streams(client, group: str, start_ms: int, end_ms: int) -> list[str]:
    """
    Return names of log streams that had events within [start_ms, end_ms].
    Capped at MAX_STREAMS to protect against pathological log groups.
    """
    streams: list[str] = []
    paginator = client.get_paginator("describe_log_streams")
    for page in paginator.paginate(
        logGroupName=group,
        orderBy="LastEventTime",
        descending=True,
    ):
        for stream in page.get("logStreams", []):
            last = stream.get("lastEventTimestamp")
            # Skip streams whose last event is older than our window.
            if last is not None and last < start_ms:
                return streams  # descending order -> nothing newer remains
            streams.append(stream["logStreamName"])
            if len(streams) >= MAX_STREAMS:
                return streams
    return streams


def fetch_stream_events(
    client, group: str, stream: str, start_ms: int, end_ms: int
) -> list[dict[str, Any]]:
    """Fetch all events for a single stream within the time window."""
    events: list[dict[str, Any]] = []
    next_token: str | None = None
    while True:
        kwargs: dict[str, Any] = {
            "logGroupName": group,
            "logStreamNames": [stream],
            "startTime": start_ms,
            "endTime": end_ms,
            "limit": 1000,
        }
        if next_token:
            kwargs["nextToken"] = next_token

        resp = client.filter_log_events(**kwargs)
        events.extend(resp.get("events", []))
        next_token = resp.get("nextToken")
        if not next_token or len(events) >= MAX_EVENTS:
            break
    return events


def fetch_all(
    client, group: str, streams: Iterable[str], start_ms: int, end_ms: int
) -> list[dict[str, Any]]:
    """Fetch events from many streams in parallel; return chronologically sorted."""
    all_events: list[dict[str, Any]] = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = {
            pool.submit(fetch_stream_events, client, group, s, start_ms, end_ms): s
            for s in streams
        }
        for fut in as_completed(futures):
            stream = futures[fut]
            try:
                all_events.extend(fut.result())
            except (BotoCoreError, ClientError) as exc:
                # Don't fail the whole run for one bad stream; warn and continue.
                print(f"warning: failed to fetch {stream}: {exc}", file=sys.stderr)
    all_events.sort(key=lambda e: e["timestamp"])
    return all_events[:MAX_EVENTS]


def format_text(events: list[dict[str, Any]]) -> str:
    """Format events as 'ISO_TIMESTAMP STREAM MESSAGE' lines."""
    lines = []
    for ev in events:
        ts = datetime.fromtimestamp(ev["timestamp"] / 1000, tz=timezone.utc).isoformat()
        # Collapse multi-line Lambda messages onto one line for greppability.
        msg = ev["message"].rstrip("\n").replace("\n", "\\n")
        lines.append(f"{ts} {ev['logStreamName']} {msg}")
    return "\n".join(lines)


def format_json(events: list[dict[str, Any]]) -> str:
    """Format events as a JSON array of normalized objects."""
    out = [
        {
            "timestamp": datetime.fromtimestamp(
                ev["timestamp"] / 1000, tz=timezone.utc
            ).isoformat(),
            "stream": ev["logStreamName"],
            "message": ev["message"],
        }
        for ev in events
    ]
    return json.dumps(out, indent=2)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Export AWS CloudWatch Logs for a relative time window."
    )
    parser.add_argument("--group", required=True, help="CloudWatch log group name")
    parser.add_argument(
        "--since",
        type=parse_since,
        default=timedelta(hours=1),
        help="Relative window, e.g. 30m, 2h, 7d (default: 1h)",
    )
    parser.add_argument(
        "--format", choices=("text", "json"), default="text", help="Output format"
    )
    parser.add_argument("--grep", help="Only include events matching this regex")
    parser.add_argument("--output", help="Write to file instead of stdout")
    parser.add_argument("--region", default=AWS_REGION, help="AWS region")
    args = parser.parse_args()

    end_dt = datetime.now(tz=timezone.utc)
    start_dt = end_dt - args.since
    start_ms = int(start_dt.timestamp() * 1000)
    end_ms = int(end_dt.timestamp() * 1000)

    try:
        client = build_client(args.region)
        streams = list_active_streams(client, args.group, start_ms, end_ms)
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code", "")
        if code == "ResourceNotFoundException":
            print(f"error: log group {args.group!r} not found in {args.region}", file=sys.stderr)
        elif code in ("AccessDeniedException", "UnauthorizedOperation"):
            print(f"error: access denied — check IAM permissions for {args.group}", file=sys.stderr)
        else:
            print(f"error: {exc}", file=sys.stderr)
        return 2
    except BotoCoreError as exc:
        print(f"error: could not reach AWS: {exc}", file=sys.stderr)
        return 2

    if not streams:
        print(f"no active streams in {args.group} for the last {args.since}", file=sys.stderr)
        return 0

    events = fetch_all(client, args.group, streams, start_ms, end_ms)

    if args.grep:
        try:
            pattern = re.compile(args.grep)
        except re.error as exc:
            print(f"error: invalid --grep regex: {exc}", file=sys.stderr)
            return 2
        events = [e for e in events if pattern.search(e["message"])]

    output = format_json(events) if args.format == "json" else format_text(events)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as fh:
            fh.write(output + "\n")
        print(f"wrote {len(events)} events to {args.output}", file=sys.stderr)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

All configuration is via environment variables (with sensible defaults) plus CLI flags:

AWS credentials are picked up via the standard chain (AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, ~/.aws/credentials, IAM role, SSO, etc.). The IAM principal needs logs:DescribeLogStreams and logs:FilterLogEvents on the target group.

Usage Examples

1. Grab the last hour of a Lambda's logs as text:

$ python3 cw_logs.py --group /aws/lambda/myfunction --since 1h
2024-05-14T10:01:22.145000+00:00 2024/05/14/[$LATEST]abc123 START RequestId: 4f2a...
2024-05-14T10:01:22.310000+00:00 2024/05/14/[$LATEST]abc123 Processing order 8821
2024-05-14T10:01:22.894000+00:00 2024/05/14/[$LATEST]abc123 END RequestId: 4f2a...

2. Filter for errors and emit JSON to a file:

$ python3 cw_logs.py --group /ecs/api --since 6h --grep 'ERROR|Traceback' --format json --output errors.json
wrote 17 events to errors.json

3. Pipe into jq for structured analysis:

$ python3 cw_logs.py --group /aws/lambda/payments --since 30m --format json | jq -r '.[].message'

Error Handling

• Missing log group → ResourceNotFoundException is caught and reported with a clear message; exit code 2.
• Insufficient IAM permissions → AccessDeniedException is detected and the user is told which group failed.
• Network / credential errors → BotoCoreError (e.g. NoCredentialsError, EndpointConnectionError) is caught at the client-build and list-streams stage.
• Per-stream failures → caught inside the thread pool; a warning goes to stderr and other streams still complete. This is deliberate: one throttled stream shouldn't nuke a 50-stream export.
• Throttling → botocore's adaptive retry mode handles ThrottlingException and RateExceeded automatically.
• Bad --since or --grep → validated up front, exit code 2 with a helpful message.
• No matching streams → prints a friendly note to stderr and exits 0 (not an error).

Extending Ideas

• Add --start/--end absolute timestamps so you can re-run the same query for a fixed historical window (useful for incident post-mortems and reproducible exports).
• Stream results instead of buffering — replace the list accumulation with a generator that yields events as they arrive, so you can tail -f-style follow large exports without holding everything in memory.
• Push to S3 or OpenSearch — swap the format_*/print step for a sink that writes NDJSON to S3 (using boto3.client("s3").put_object) or bulk-indexes into OpenSearch for long-term log search.
• Add CloudWatch Logs Insights support — accept a --query flag and call start_query/get_query_results for server-side filtering, which is far cheaper and faster than filter_log_events for large groups.