scripts/aws/cloudwatch-log-costs.py

#!/usr/bin/env python3
#
# Determine which CloudWatch log groups are contributing most cost and
# estimate total monthly cost for an AWS account.
#
# Based on: https://repost.aws/knowledge-center/cloudwatch-logs-bill-increase

import boto3
from datetime import datetime, timedelta

cw = boto3.client("cloudwatch")

# We use a period of 14 days because the ListMetrics call returns log groups
# that ingested data in the last 14 days.
period = timedelta(days=14)
end_time = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_time = end_time - period

n = 0
query_names = {}
queries = []

paginator = cw.get_paginator("list_metrics")
for page in paginator.paginate(Namespace="AWS/Logs", MetricName="IncomingBytes"):
    for metric in page["Metrics"]:
        if not metric["Dimensions"]:
            continue
        query = {
            "Id": f"q{n}",
            "MetricStat": {
                "Metric": {
                    "Namespace": metric["Namespace"],
                    "MetricName": metric["MetricName"],
                    "Dimensions": metric["Dimensions"],
                },
                "Period": int(period.total_seconds()),
                "Stat": "Sum",
            },
        }
        query_names[f"q{n}"] = metric["Dimensions"][0]["Value"]
        n = n + 1
        queries.append(query)


# The GetMetricData query can retrieve up to 500 metrics in a single request.
# The following code assumes we don't have more than 500 active log groups,
# otherwise we would have to chunk up the calls to GetMetricData.
results = []
data_paginator = cw.get_paginator("get_metric_data")
for data_page in data_paginator.paginate(
    MetricDataQueries=queries, StartTime=start_time, EndTime=end_time
):
    for result in data_page["MetricDataResults"]:
        if result["Values"]:
            results.append(
                {
                    "LogGroup": query_names[result["Id"]],
                    "IncomingBytes": result["Values"][0],
                }
            )

results.sort(key=lambda x: x["IncomingBytes"], reverse=True)

incoming_bytes_total = 0
for result in results:
    incoming_bytes_total += result["IncomingBytes"]
    print("{LogGroup}: {IncomingBytes}".format(**result))

# Estimate a monthly cost by doubling the incoming_bytes counted over the last 14 days
incoming_gb = 2 * incoming_bytes_total / (1024 * 1024 * 1024)
# CloudWatch logs: 5GiB free ingestion then $0.50 / GiB
cost = (incoming_gb - 5.0) * 0.5
print(f"Estimated monthly cost: ${cost:.2f}")