Build a CloudWatch Logs Cost Agent: Find Never-Expire Log Groups, Ingestion Hot Spots, and Debug Floods Before the Bill Does

aws ce get-cost-and-usage \
  --time-period Start=2026-08-01,End=2026-09-01 --granularity MONTHLY \
  --metrics UnblendedCost \
  --filter '{"Dimensions":{"Key":"SERVICE","Values":["AmazonCloudWatch"]}}' \
  --group-by Type=DIMENSION,Key=USAGE_TYPE \
  --query 'ResultsByTime[0].Groups[?contains(Keys[0], `Bytes`) || contains(Keys[0], `ByteHrs`)].[Keys[0], Metrics.UnblendedCost.Amount]' \
  --output table

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadLogGroupsAndMetrics",
      "Effect": "Allow",
      "Action": [
        "logs:DescribeLogGroups", "logs:DescribeSubscriptionFilters",
        "logs:DescribeMetricFilters", "logs:ListTagsForResource",
        "logs:StartQuery", "logs:GetQueryResults", "logs:StopQuery",
        "cloudwatch:GetMetricData", "ce:GetCostAndUsage",
        "lambda:GetFunctionConfiguration", "lambda:ListFunctions"
      ],
      "Resource": "*"
    },
    {
      "Sid": "NeverEvenIfSomeoneAddsIt",
      "Effect": "Deny",
      "Action": [
        "logs:DeleteLogGroup", "logs:DeleteLogStream", "logs:PutRetentionPolicy",
        "logs:DeleteRetentionPolicy", "logs:PutSubscriptionFilter",
        "logs:DeleteSubscriptionFilter", "logs:CreateLogGroup"
      ],
      "Resource": "*"
    }
  ]
}

import boto3
from datetime import datetime, timedelta, timezone

logs = boto3.client("logs")
cw = boto3.client("cloudwatch")
NOW = datetime.now(timezone.utc)
INGEST_PRICE = {"STANDARD": 0.50, "INFREQUENT_ACCESS": 0.25}
STORAGE_PRICE = 0.03

def all_log_groups():
    for page in logs.get_paginator("describe_log_groups").paginate():
        yield from page["logGroups"]

def incoming_bytes(names, days=30):
    """Sum of IncomingBytes per group over the window, 500 groups per call."""
    out = {}
    for i in range(0, len(names), 500):
        chunk = names[i:i + 500]
        queries = [{
            "Id": f"q{j}",
            "MetricStat": {
                "Metric": {"Namespace": "AWS/Logs", "MetricName": "IncomingBytes",
                           "Dimensions": [{"Name": "LogGroupName", "Value": n}]},
                "Period": days * 86400, "Stat": "Sum"},
        } for j, n in enumerate(chunk)]
        resp = cw.get_metric_data(MetricDataQueries=queries,
                                  StartTime=NOW - timedelta(days=days), EndTime=NOW)
        for r in resp["MetricDataResults"]:
            out[chunk[int(r["Id"][1:])]] = sum(r["Values"])
    return out

def source_of(name):
    rules = [("/aws/lambda/", "lambda"), ("/aws/eks/", "eks_control_plane"),
             ("/aws/containerinsights/", "container_insights"), ("/aws/rds/", "rds"),
             ("/aws/vpc/", "vended_flow_logs"), ("API-Gateway-Execution-Logs", "api_gateway"),
             ("/aws/codebuild/", "codebuild"), ("/aws/cloudtrail", "audit"), ("/ecs/", "ecs")]
    return next((label for prefix, label in rules if name.startswith(prefix)), "application")

def evidence(g, ingest):
    name = g["logGroupName"]
    cls = g.get("logGroupClass", "STANDARD")
    gb_in = ingest.get(name, 0.0) / 1e9
    stored_gb = g.get("storedBytes", 0) / 1e9
    arn = g.get("logGroupArn") or g["arn"].removesuffix(":*")
    subs = logs.describe_subscription_filters(logGroupName=name)["subscriptionFilters"]
    return {
        "name": name, "source": source_of(name), "class": cls,
        "retention_days": g.get("retentionInDays"),      # None means never expire
        "created": datetime.fromtimestamp(g["creationTime"] / 1000, timezone.utc).date().isoformat(),
        "stored_gb": round(stored_gb, 2),
        "ingest_gb_30d": round(gb_in, 2),
        "ingest_usd_30d": round(gb_in * INGEST_PRICE[cls], 2),
        "storage_usd_month": round(stored_gb * STORAGE_PRICE, 2),
        "subscriptions": [s["destinationArn"] for s in subs],
        "metric_filters": len(logs.describe_metric_filters(logGroupName=name)["metricFilters"]),
        "tags": logs.list_tags_for_resource(resourceArn=arn).get("tags", {}),
    }

import time

def debug_share(name, hours=1):
    q = logs.start_query(
        logGroupName=name,
        startTime=int((NOW - timedelta(hours=hours)).timestamp()), endTime=int(NOW.timestamp()),
        queryString='stats count(*) as total, '
                    'sum(strcontains(@message, "DEBUG")) as debug, '
                    'sum(strcontains(@message, "TRACE")) as trace')
    while (r := logs.get_query_results(queryId=q["queryId"]))["status"] in ("Scheduled", "Running"):
        time.sleep(1)
    row = {f["field"]: float(f["value"]) for f in r["results"][0]} if r["results"] else {}
    total = row.get("total", 0) or 1
    return round((row.get("debug", 0) + row.get("trace", 0)) / total, 3)

{
  "name": "propose_log_group_actions",
  "description": "Propose one action per log group from the evidence records provided.",
  "input_schema": {
    "type": "object",
    "properties": {
      "proposals": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "log_group": {"type": "string"},
            "classification": {"enum": ["lambda_default", "debug_flood", "vended_flow_logs",
                                         "container_insights", "eks_control_plane", "audit",
                                         "application", "unknown"]},
            "action": {"enum": ["set_retention", "reduce_log_level",
                                 "recreate_as_infrequent_access", "route_to_s3", "no_change"]},
            "retention_days": {"enum": [1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365,
                                         400, 545, 731, 1096, 1827, 2192, 2557, 2922, 3288, 3653, null]},
            "monthly_saving_usd": {"type": "number"},
            "saving_requires": {"enum": ["nothing", "retention_ageout", "code_or_config_change"]},
            "rationale": {"type": "string"},
            "confidence": {"enum": ["high", "medium", "low"]}
          },
          "required": ["log_group", "classification", "action", "monthly_saving_usd",
                       "saving_requires", "rationale", "confidence"]
        }
      }
    },
    "required": ["proposals"]
  }
}

You are a FinOps reviewer for CloudWatch Logs. You receive evidence records and
propose actions. Rules:
1. Never propose retention below the "retention:min" tag if present, and never
   below 365 for classification "audit" or any group whose name contains
   "cloudtrail", "audit", or "security". Set action=no_change and say why.
2. Ingestion savings (reduce_log_level, route_to_s3, recreate_as_infrequent_access)
   require a change by the owning team. Set saving_requires=code_or_config_change
   and never count them as realised.
3. Retention savings are storage only: stored_gb * 0.03 per month, realised
   after the old data ages out. Do not add ingestion cost to that number.
4. recreate_as_infrequent_access is only valid when metric_filters == 0 and
   subscriptions is empty; the IA class supports neither.
5. A group with a subscription to a SIEM or S3 destination already has a copy
   elsewhere; 30 days retention is the default proposal there.
6. If tags carry no owner and the name gives no clue, classification=unknown,
   action=no_change, confidence=low. Do not guess an owner.
Explain every proposal in one sentence a service owner would accept.

import {
  to = aws_cloudwatch_log_group.lambda_thumbnailer
  id = "/aws/lambda/thumbnailer"
}

resource "aws_cloudwatch_log_group" "lambda_thumbnailer" {
  name              = "/aws/lambda/thumbnailer"
  retention_in_days = 30
}

terraform show -json plan.bin | jq -e '
  [.resource_changes[]
   | select(.change.actions != ["no-op"])
   | select(.type != "aws_cloudwatch_log_group" or .change.actions != ["update"])]
  | length == 0' || { echo "bot PR may only update existing log groups"; exit 1; }

resource "aws_lambda_function" "thumbnailer" {
  # ...existing config...
  logging_config {
    log_format            = "JSON"
    application_log_level = "INFO"
    system_log_level      = "WARN"
    log_group             = aws_cloudwatch_log_group.lambda_thumbnailer.name
  }
}

{
  "source": ["aws.logs"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventSource": ["logs.amazonaws.com"],
    "eventName": ["CreateLogGroup"]
  }
}

import boto3
logs = boto3.client("logs")

def handler(event, _):
    name = event["detail"]["requestParameters"]["logGroupName"]
    groups = logs.describe_log_groups(logGroupNamePrefix=name)["logGroups"]
    g = next((x for x in groups if x["logGroupName"] == name), None)
    if g and g.get("retentionInDays") is None:
        logs.put_retention_policy(logGroupName=name, retentionInDays=30)