Amazon CloudWatch — Logs, Logs Insights & Metric Filters¶

logs.create_log_group(logGroupName="/qa/orders", logGroupClass="STANDARD", tags={"env": "test"})
logs.put_retention_policy(logGroupName="/qa/orders", retentionInDays=14)    # delete_retention_policy = never expire

# app/logs.py (excerpt)
def groups_without_retention(client, prefix=None) -> list[str]:
    kwargs = {"logGroupNamePrefix": prefix} if prefix else {}
    pages = client.get_paginator("describe_log_groups").paginate(**kwargs)
    return sorted(g["logGroupName"] for page in pages for g in page["logGroups"] if "retentionInDays" not in g)

# app/jsonlog.py
_RESERVED = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"}

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(timespec="milliseconds"),
            "level": record.levelname, "logger": record.name, "message": record.getMessage(),
        }
        payload.update({k: v for k, v in record.__dict__.items() if k not in _RESERVED})   # extra={...} fields
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)    # the traceback stays inside one event
        return json.dumps(payload, default=str)

def configure(level=logging.INFO, stream=None) -> None:
    handler = logging.StreamHandler(stream or sys.stdout)       # stdout: the awslogs driver and the agent read it
    handler.setFormatter(JsonFormatter())
    logging.getLogger().handlers[:] = [handler]
    logging.getLogger().setLevel(level)

log.warning("order %s failed", order_id, extra={"order_id": order_id, "latency_ms": 412, "request_id": rid})

{"timestamp": "2026-10-22T12:00:03.114+00:00", "level": "WARNING", "logger": "orders", "message": "order o-1 failed", "order_id": "o-1", "latency_ms": 412, "request_id": "r-77"}

"logConfiguration": {
  "logDriver": "awslogs",
  "options": {"awslogs-create-group": "true", "awslogs-group": "/ecs/orders-api",
              "awslogs-region": "eu-west-1", "awslogs-stream-prefix": "api",
              "mode": "non-blocking", "max-buffer-size": "25m"}
}

{
  "metrics": {"append_dimensions": {"InstanceId": "${aws:InstanceId}"},
              "metrics_collected": {"mem": {"measurement": ["used_percent"]}, "disk": {"measurement": ["used_percent"]}}},
  "logs": {"logs_collected": {"files": {"collect_list": [
    {"file_path": "/var/log/orders/app.log", "log_group_name": "/ec2/orders",
     "log_stream_name": "{instance_id}", "retention_in_days": 14}]}}}
}

# app/logs.py (excerpt)
def batches(events):
    """Split events (sorted here) into PutLogEvents batches that respect size, count and time-span limits."""
    current, size = [], 0
    for event in sorted(events, key=lambda e: e["timestamp"]):
        if current and (size + event_size(event) > MAX_BATCH_BYTES or len(current) >= MAX_BATCH_EVENTS
                        or event["timestamp"] - current[0]["timestamp"] > MAX_SPAN_MS):
            yield current
            current, size = [], 0
        current.append(event)
        size += event_size(event)
    if current:
        yield current

def put_events(client, group, stream, events) -> int:
    sent = 0
    for batch in batches(events):
        response = client.put_log_events(logGroupName=group, logStreamName=stream, logEvents=batch)
        if response.get("rejectedLogEventsInfo"):
            raise RejectedEvents(response["rejectedLogEventsInfo"])      # HTTP 200 does not mean accepted
        sent += len(batch)
    return sent

start = int((time.time() - 3600) * 1000)                                     # Logs APIs take MILLISECONDS
pages = logs.get_paginator("filter_log_events").paginate(
    logGroupName="/ecs/orders-api", filterPattern='{ $.level = "ERROR" }', startTime=start)
errors = [e["message"] for page in pages for e in page["events"]]


def read_stream(logs, group, stream):
    """get_log_events has no paginator: follow nextForwardToken until the service returns the token you sent."""
    events, token = [], None
    while True:
        kwargs = {"logGroupName": group, "logStreamName": stream, "startFromHead": True}
        if token:
            kwargs["nextToken"] = token
        page = logs.get_log_events(**kwargs)
        events += page["events"]
        if page["nextForwardToken"] == token:
            return events
        token = page["nextForwardToken"]

# app/logs.py (excerpt)
def run_query(client, groups, query, start, end, *, timeout=60, interval=1.0, sleep=time.sleep, clock=time.monotonic):
    query_id = client.start_query(logGroupNames=groups, startTime=int(start.timestamp()),      # SECONDS here
                                  endTime=int(end.timestamp()), queryString=query)["queryId"]
    deadline = clock() + timeout
    while True:
        response = client.get_query_results(queryId=query_id)
        if response["status"] == "Complete":
            return [{f["field"]: f["value"] for f in row if f["field"] != "@ptr"} for row in response["results"]]
        if response["status"] in {"Failed", "Cancelled", "Timeout", "Unknown"}:
            raise QueryFailed(response["status"])
        if clock() >= deadline:                                  # Scheduled or Running for too long
            client.stop_query(queryId=query_id)
            raise TimeoutError(f"query {query_id} still {response['status']} after {timeout}s")
        sleep(interval)

# 1. Errors per 5 minutes (time series)
filter level = "ERROR" | stats count(*) as errors by bin(5m)

# 2. Latest 20 errors with context
fields @timestamp, message, order_id, request_id | filter level = "ERROR" | sort @timestamp desc | limit 20

# 3. Latency percentiles per route, worst first
filter ispresent(latency_ms) | stats count(*) as n, avg(latency_ms) as avg_ms, pct(latency_ms, 50) as p50, pct(latency_ms, 95) as p95, pct(latency_ms, 99) as p99 by route | sort p99 desc

# 4. Most frequent error messages
filter level = "ERROR" | stats count(*) as n by message | sort n desc | limit 10

# 5. Follow one request across streams
filter request_id = "r-77" | fields @timestamp, @logStream, level, message | sort @timestamp asc

# 6. Extract fields from unstructured lines with parse
parse @message "user=* action=*" as user, action | stats count(*) as n by action | sort n desc

# 7. How many distinct orders were touched
filter ispresent(order_id) | stats countDistinct(order_id) as orders

# 8. Where does the ingestion volume come from (cost)
fields strlen(@message) as bytes | stats sum(bytes) / 1024 / 1024 as mb by @logStream | sort mb desc | limit 10

logs.put_metric_filter(
    logGroupName="/ecs/orders-api", filterName="errors", filterPattern='{ $.level = "ERROR" }',
    metricTransformations=[{"metricName": "ErrorCount", "metricNamespace": "QA/Orders",
                            "metricValue": "1",        # or "$.latency_ms" for JSON, "$size" for space-delimited
                            "defaultValue": 0,         # emitted when the pattern does not match, so the metric is not empty
                            "unit": "Count"}])

logs.put_subscription_filter(
    logGroupName="/ecs/orders-api", filterName="to-firehose", filterPattern="",
    destinationArn="arn:aws:firehose:eu-west-1:123456789012:deliverystream/logs",
    roleArn="arn:aws:iam::123456789012:role/cwl-to-firehose")