AWS CloudWatch Logs : Log Group ごとのログ送信量を確認する

CloudWatch Logs の料金

CloudWatch Logs へログを保存すると、料金は以下のように計算されます。(単価は 2025/01 時点の東京リージョン)

参考) Amazon CloudWatch Pricing

• Collect (Data Ingestion) : 0.76 USD/GB ※Infrequent Access の場合は半額
• Store (Archival) : 0.033 USD/GB･月 ※圧縮後のサイズ
• その他 (参照、分析時)

このうち意外とばかにならないのがログ送信時にかかる Data Ingestion のコストです。

メトリクス

ロググループごとの送信量は以下の CloudWatch Metrics に記録されています。

• Namespace: AWS/Logs
• MetricName: IncomingBytes
• Dimensions
  • LogGroupName: (ロググループ名)

とあるデータベースの監査ログが月あたり約 500 GB 程度発生しており、400 USD/月ものコストがかかっていることが分かります。

一括確認

以下のスクリプトで、指定した期間の各ロググループの合計バイト数を確認することができます。

#!/usr/bin/env python

import argparse
from datetime import datetime
from typing import Optional, Iterable

import boto3
from botocore.config import Config


def get_logs_incoming_bytes(region_name: Optional[str], log_group_name: str,
                            start_datetime: datetime, end_datetime: datetime) -> int:
    client = boto3.client("cloudwatch", region_name=region_name, config=Config(retries={"mode": "standard"}))
    response = client.get_metric_data(
        MetricDataQueries=[
            {
                "Id": "incomingBytes",
                "MetricStat": {
                    "Metric": {
                        "Namespace": "AWS/Logs",
                        "MetricName": "IncomingBytes",
                        "Dimensions": [
                            {
                                "Name": "LogGroupName",
                                "Value": f"{log_group_name}"
                            }
                        ]
                    },
                    "Period": 3600,  # per day
                    "Stat": "Sum"
                },
                "ReturnData": True
            },
        ],
        StartTime=start_datetime,
        EndTime=end_datetime,
    )
    values = response["MetricDataResults"][0]["Values"]
    total = 0
    for value in values:
        total += value
    return int(total)


def get_log_group_names() -> Iterable[str]:
    logs_client = boto3.client("logs", config=Config(retries={"mode": "standard"}))
    params = {}
    while True:
        res = logs_client.describe_log_groups(**params)
        for log_group in res["logGroups"]:
            yield log_group["logGroupName"]
        if "nextToken" not in res:
            break
        params["nextToken"] = res["nextToken"]


def run(region_name: Optional[str], log_group_name: Optional[str],
        start_datetime: datetime, end_datetime: datetime):
    if log_group_name:
        log_group_names = [log_group_name]
    else:
        log_group_names = get_log_group_names()
    for log_group_name in log_group_names:
        value = get_logs_incoming_bytes(region_name, log_group_name, start_datetime, end_datetime)
        print(f"{value},{log_group_name}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--region")
    parser.add_argument("--log-group-name")
    parser.add_argument("--start", required=True)
    parser.add_argument("--end", required=True)
    args = parser.parse_args()

    start_ts = datetime.fromisoformat(args.start)
    end_ts = datetime.fromisoformat(args.end)
    run(args.region, args.log_group_name, start_ts, end_ts)


if __name__ == "__main__":
    main()