AWS Config の定期的記録と連続的記録どちらが安くなるのか比較してみた

AWS Config の定期的記録と連続的記録どちらが安くなるのか比較してみた

Clock Icon2025.01.19

こんにちは!クラウド事業本部コンサルティング部のたかくに(@takakuni_)です。

みなさん AWS Config 使ってますでしょうか。

AWS Config にはリソースの記録頻度に、変更の都度記録する連続的な記録と日時で記録する定期的な記録の 2 パターン存在します。

https://dev.classmethod.jp/articles/reinvent2023-updata-aws-config-periodic-recording/

各取得頻度に発生するコストは異なり、コスト最適化観点のみでコメントすると 1 日 4 回以上変更が発生するケースでは定期的な記録に変更することでコストメリットが生まれてきます。(逆に変更量が 1 日 4 回より小さい頻度で定期的な記録を利用しているとコスト増につながります。)

各 AWS リージョンの AWS アカウントごとに配信される設定項目あたりのコスト

  • 連続的な記録:USD 0.003
  • 定期的な記録:USD 0.012

https://aws.amazon.com/jp/config/pricing/

今回はどちらの方がコスト的に良いのか、リソース別に判定するチェックツールを作成してみます。

前提

今回は AWS Config のメトリクスをベースに集計します。

すでに定期的な記録を実施している場合、メトリクスは 1 日 1 回取得され損益分岐を計算できないため、今回は連続的な記録を利用していて、 AWS Config の利用費が目立つケースに特定します。

やってみた

作成したコードが以下になります。

main.py
#!/usr/bin/env python3
"""
AWS Config のリソースタイプごとの記録方式による損益を計算するスクリプト
"""

import logging
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import boto3
from botocore.exceptions import BotoCoreError, ClientError

# 定数定義
class Constants:
    CONTINUOUS_RECORDING_PRICE = 0.003
    PERIODIC_RECORDING_PRICE = 0.012
    DAYS_TO_ANALYZE = 30
    SECONDS_PER_DAY = 86400
    AWS_REGION = 'ap-northeast-1'

# ログ設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class ConfigCostCalculator:
    def __init__(self):
        """初期化"""
        try:
            self.cw = boto3.client('cloudwatch', region_name=Constants.AWS_REGION)
        except Exception as e:
            logger.error(f"Failed to initialize AWS client: {e}")
            raise

    def list_metrics(self) -> List[str]:
        """
        Config に記録されているリソースタイプを取得

        Returns:
            List[str]: リソースタイプのリスト
        """
        try:
            response = self.cw.list_metrics(
                Namespace='AWS/Config',
                MetricName='ConfigurationItemsRecorded'
            )

            resource_types = [
                dimension['Value']
                for metric in response['Metrics']
                for dimension in metric['Dimensions']
                if dimension['Name'] == 'ResourceType'
            ]

            logger.info(f"Found {len(resource_types)} resource types")
            return resource_types

        except (BotoCoreError, ClientError) as e:
            logger.error(f"AWS API error while listing metrics: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error while listing metrics: {e}")
            raise

    def get_metric_statistics(self, resource_type: str) -> List[Dict]:
        """
        リソースタイプごとのメトリクスを取得

        Args:
            resource_type (str): リソースタイプ

        Returns:
            List[Dict]: メトリクスデータポイントのリスト
        """
        try:
            end_time = datetime.now()
            start_time = end_time - timedelta(days=Constants.DAYS_TO_ANALYZE)

            response = self.cw.get_metric_statistics(
                Namespace='AWS/Config',
                MetricName='ConfigurationItemsRecorded',
                Dimensions=[
                    {
                        'Name': 'ResourceType',
                        'Value': resource_type
                    }
                ],
                StartTime=start_time,
                EndTime=end_time,
                Period=Constants.SECONDS_PER_DAY,
                Statistics=['Sum']
            )

            data_points = response['Datapoints']
            logger.debug(f"Retrieved {len(data_points)} data points for {resource_type}")
            return data_points

        except (BotoCoreError, ClientError) as e:
            logger.error(f"AWS API error while getting metrics for {resource_type}: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error while getting metrics for {resource_type}: {e}")
            raise

    def calculate_profit_and_loss(self, data_points: List[Dict]) -> float:
        """
        損益を計算

        Args:
            data_points (List[Dict]): メトリクスデータポイント

        Returns:
            float: 計算された損益
        """
        try:
            total = 0.0

            # データポイントがない日の定期的な記録のコストを計算
            days_without_data = Constants.DAYS_TO_ANALYZE - len(data_points)
            cost_of_periodic_recording_without_datapoints = (
                days_without_data * Constants.PERIODIC_RECORDING_PRICE
            )
            total += cost_of_periodic_recording_without_datapoints

            # データポイントがある日の連続記録と定期的な記録の損益を計算
            for data_point in data_points:
                cost_of_periodic_recording = Constants.PERIODIC_RECORDING_PRICE
                cost_of_continuous_recording = (
                    data_point['Sum'] * Constants.CONTINUOUS_RECORDING_PRICE
                )
                total += cost_of_periodic_recording - cost_of_continuous_recording

            return total

        except Exception as e:
            logger.error(f"Error calculating profit and loss: {e}")
            raise

    def process_resource_type(self, resource_type: str) -> Optional[float]:
        """
        リソースタイプごとの処理

        Args:
            resource_type (str): リソースタイプ

        Returns:
            Optional[float]: 計算された損益、エラー時はNone
        """
        try:
            if resource_type == 'All':
                return None

            data_points = self.get_metric_statistics(resource_type)
            return self.calculate_profit_and_loss(data_points)

        except Exception as e:
            logger.error(f"Error processing resource type {resource_type}: {e}")
            return None

def main():
    """メイン処理"""
    try:
        calculator = ConfigCostCalculator()
        resource_types = calculator.list_metrics()

        if not resource_types:
            logger.warning("No resource types found")
            return

        print("\n定期的な記録に変更した場合のコスト削減額:")
        print("-" * 50)

        for resource_type in resource_types:
            result = calculator.process_resource_type(resource_type)
            if result is not None:
                print(f"{resource_type}: {result:.2f} USD")

        print("-" * 50)

    except Exception as e:
        logger.error(f"Application error: {e}")
        raise

if __name__ == '__main__':
    try:
        main()
    except Exception as e:
        logger.critical(f"Critical error: {e}")
        exit(1)

CloudWatch メトリクスの取得

AWS Config は記録されたリソースのみメトリクスを発行するため、闇雲にリソースタイプから引っ張るのではなく、まずは何が記録されているのかを知ることが大事です。そこで、 list_metrics を定義しリソースタイプの取得を行います。

    def list_metrics(self) -> List[str]:
        """
        Config に記録されているリソースタイプを取得

        Returns:
            List[str]: リソースタイプのリスト
        """
        try:
            response = self.cw.list_metrics(
                Namespace='AWS/Config',
                MetricName='ConfigurationItemsRecorded'
            )

            resource_types = [
                dimension['Value']
                for metric in response['Metrics']
                for dimension in metric['Dimensions']
                if dimension['Name'] == 'ResourceType'
            ]

            logger.info(f"Found {len(resource_types)} resource types")
            return resource_types

        except (BotoCoreError, ClientError) as e:
            logger.error(f"AWS API error while listing metrics: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error while listing metrics: {e}")
            raise

リソースタイプ別のメトリクス取得

プロダクトのフェーズによって変更量は異なると思いますが、今回は集計期間を 30 日としました。

取得したリソースタイプに対して、 for ループを回し、 1 日単位の 30 日間のメトリクス取得を行います。Constants クラスの DAYS_TO_ANALYZE で集計期間を変更可能にしているため、各々の設定したい範囲に変更してください。

    def get_metric_statistics(self, resource_type: str) -> List[Dict]:
        """
        リソースタイプごとのメトリクスを取得

        Args:
            resource_type (str): リソースタイプ

        Returns:
            List[Dict]: メトリクスデータポイントのリスト
        """
        try:
            end_time = datetime.now()
            start_time = end_time - timedelta(days=Constants.DAYS_TO_ANALYZE)

            response = self.cw.get_metric_statistics(
                Namespace='AWS/Config',
                MetricName='ConfigurationItemsRecorded',
                Dimensions=[
                    {
                        'Name': 'ResourceType',
                        'Value': resource_type
                    }
                ],
                StartTime=start_time,
                EndTime=end_time,
                Period=Constants.SECONDS_PER_DAY,
                Statistics=['Sum']
            )

            data_points = response['Datapoints']
            logger.debug(f"Retrieved {len(data_points)} data points for {resource_type}")
            return data_points

        except (BotoCoreError, ClientError) as e:
            logger.error(f"AWS API error while getting metrics for {resource_type}: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error while getting metrics for {resource_type}: {e}")
            raise

損益分析

まず、データポイントがない日(リソースが変更されていない)を計算し、定期的な記録の費用を追加します。

例:2 日しかない場合は (30 - 2) * 0.012

続いて、データポイントがある日に関しては、定期的な記録の費用から連続的な記録の費用をマイナスした結果を合計に足すことで、月間ベースでの損益分析を行ないました。

    def calculate_profit_and_loss(self, data_points: List[Dict]) -> float:
        """
        損益を計算

        Args:
            data_points (List[Dict]): メトリクスデータポイント

        Returns:
            float: 計算された損益
        """
        try:
            total = 0.0

            # データポイントがない日の定期的な記録のコストを計算
            days_without_data = Constants.DAYS_TO_ANALYZE - len(data_points)
            cost_of_periodic_recording_without_datapoints = (
                days_without_data * Constants.PERIODIC_RECORDING_PRICE
            )
            total += cost_of_periodic_recording_without_datapoints

            # データポイントがある日の連続記録と定期的な記録の損益を計算
            for data_point in data_points:
                cost_of_periodic_recording = Constants.PERIODIC_RECORDING_PRICE
                cost_of_continuous_recording = (
                    data_point['Sum'] * Constants.CONTINUOUS_RECORDING_PRICE
                )
                total += cost_of_periodic_recording - cost_of_continuous_recording

            return total

        except Exception as e:
            logger.error(f"Error calculating profit and loss: {e}")
            raise

実行してみた

それでは実行してみましょう。

ある環境では ECS のデプロイが頻繁に行われているため、 ENI や Security Group の料金が、若干ではありますが削減できますね。

takakuni % python main.py
2025-01-19 17:49:47,910 - INFO - Found 15 resource types

定期的な記録に変更した場合のコスト削減額:
--------------------------------------------------
AWS::CloudWatch::Alarm: -0.04 USD
AWS::ECS::TaskDefinition: -0.17 USD
AWS::ECS::Service: 0.01 USD
AWS::IAM::Policy: 0.35 USD
AWS::EC2::SecurityGroup: -0.99 USD
AWS::EC2::Subnet: -1.26 USD
AWS::RDS::DBClusterSnapshot: 0.01 USD
AWS::EC2::NetworkInterface: -2.10 USD
AWS::IAM::Role: 0.33 USD
AWS::WAFv2::WebACL: 0.36 USD
AWS::S3::Bucket: 0.29 USD
AWS::EC2::VPNConnection: 0.33 USD
AWS::EC2::VPC: -0.95 USD
AWS::Config::ResourceCompliance: -0.36 USD
--------------------------------------------------

また、ある環境では、変更量が多くないためコスト増につながっていますね。

(-0.00USD は小数点 3 桁以下でマイナスになってて、私が変換サボっているだけです。)

takakuni % python main.py
2025-01-19 17:51:15,891 - INFO - Found 21 resource types

定期的な記録に変更した場合のコスト削減額:
--------------------------------------------------
AWS::CloudFormation::Stack: 0.20 USD
AWS::EC2::InternetGateway: 0.31 USD
AWS::EC2::NatGateway: 0.21 USD
AWS::EC2::RouteTable: 0.12 USD
AWS::EC2::VPC: 0.10 USD
AWS::EC2::NetworkAcl: 0.30 USD
AWS::EC2::Subnet: -0.00 USD
AWS::EC2::SecurityGroup: 0.02 USD
AWS::EC2::EIP: 0.26 USD
AWS::EC2::VPCEndpoint: 0.32 USD
AWS::EKS::Cluster: 0.28 USD
AWS::EC2::SubnetRouteTableAssociation: 0.23 USD
AWS::EKS::Addon: 0.25 USD
AWS::S3::Bucket: 0.23 USD
AWS::StepFunctions::StateMachine: 0.35 USD
AWS::Amplify::App: 0.36 USD
AWS::IAM::Policy: 0.02 USD
AWS::DynamoDB::Table: 0.33 USD
AWS::Route53Resolver::ResolverRuleAssociation: 0.34 USD
AWS::IAM::Role: -0.38 USD
--------------------------------------------------

まとめ

以上、「AWS Config の定期的記録と連続的記録どちらが安くなるのか比較してみた」でした。

変更量の大きい環境だと、地味に痛い Config の料金。

定期的な変更を記録する方針で OK の場合は、リソースタイプ別に設定変更を検討いただくのが良いかと思います。

このブログがどなたかの参考になれば幸いです。

クラウド事業本部コンサルティング部のたかくに(@takakuni_)でした!

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.