【ほぼ自分用】RDSのAutoScallingをクラスターDBに紐づくリードレプリカのいづれか一つがCPU使用率の閾値を超えた場合に発動させるLambda

目次

1.はじめに
2.前提条件
3.実装内容について
4.Lambda関数の中身
5.まとめ

1. はじめに

AWSコンソール上ではクラスター全体のCPU使用率の平均値がトリガーのAutoScallingとなってしまうため、CLIで設定できるクラスターに紐づいたリードレプリカ各々のCPU使用率をトリガーとしてAutoScallingさせたい旨で開発しました。
この要件実装したいよって人が他にも現れたら役に立てばなと

2. 前提条件

前提条件として、Lambda関数実行用のIAMロールが必要です。
RDS制御系のポリシーとCloudwatchにログを出力するための許可ポリシーあたりをアタッチしておけば問題ありません。
今回は、手っ取り早く以下二つを使用しました。
・AmazonRDSFullAccess
・CloudWatchFullAccess

3. 実装内容について

Lambdaで実装します。
ランタイムはpython3.12を使用しています。
SDKはboto3
必要な情報は環境変数に入れる値です。
以下を参照くださいませ。

Key	value（値）
DB_CLUSTER_IDENTIFIER	クラスター名
DYNAMIC_SCALING_IN_SUSPENDED	FALSEorTRUE
DYNAMIC_SCALING_OUT_SUSPENDED	FALSEorTRUE
SCHEDULED_SCALING_SUSPENDED	FALSEorTRUE
MAX_CAPACITY	1~15
MIN_CAPACITY	1~15
POLICY_NAME	設定するポリシー名
SCALE_IN_COOLDOWN	1800
SCALE_OUT_COOLDOWN	1800
TARGET_VALUE	45

4. Lambda関数の中身

AppuricationAutoScalligSetting

import boto3
import os
import logging

# AWSクライアントの初期化
app_scaling_client = boto3.client('application-autoscaling')

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def load_environment_variables():
    """環境変数から設定を読み込む"""
    return {
        'db_cluster_identifier': os.environ['DB_CLUSTER_IDENTIFIER'],
        'min_capacity': int(os.environ['MIN_CAPACITY']),
        'max_capacity': int(os.environ['MAX_CAPACITY']),
        'policy_name': os.environ['POLICY_NAME'],
        'target_value': float(os.environ['TARGET_VALUE']),
        'scale_in_cooldown': int(os.environ['SCALE_IN_COOLDOWN']),
        'scale_out_cooldown': int(os.environ['SCALE_OUT_COOLDOWN']),
        'dynamic_scaling_out_suspended': os.environ['DYNAMIC_SCALING_OUT_SUSPENDED'] == 'TRUE',
        'dynamic_scaling_in_suspended': os.environ['DYNAMIC_SCALING_IN_SUSPENDED'] == 'TRUE',
        'scheduled_scaling_suspended': os.environ['SCHEDULED_SCALING_SUSPENDED'] == 'TRUE'
    }

def register_scalable_target(settings):
    """スケーラブルターゲットの登録または再登録"""
    try:
        app_scaling_client.register_scalable_target(
            ServiceNamespace='rds',
            ResourceId=f'cluster:{settings["db_cluster_identifier"]}',
            ScalableDimension='rds:cluster:ReadReplicaCount',
            MinCapacity=settings['min_capacity'],
            MaxCapacity=settings['max_capacity'],
            SuspendedState={
                'DynamicScalingInSuspended': settings['dynamic_scaling_in_suspended'],
                'DynamicScalingOutSuspended': settings['dynamic_scaling_out_suspended'],
                'ScheduledScalingSuspended': settings['scheduled_scaling_suspended']
            }
        )
        logger.info("スケーラブルターゲットの登録に成功しました。")
    except Exception as e:
        logger.error(f"スケーラブルターゲットの登録に失敗しました。エラー: {e}")

def update_scaling_policy(settings):
    """スケーリングポリシーの設定または更新"""
    try:
        app_scaling_client.put_scaling_policy(
            PolicyName=settings['policy_name'],
            ServiceNamespace='rds',
            ResourceId=f'cluster:{settings["db_cluster_identifier"]}',
            ScalableDimension='rds:cluster:ReadReplicaCount',
            PolicyType='TargetTrackingScaling',
            TargetTrackingScalingPolicyConfiguration={
                'TargetValue': settings['target_value'],
                'CustomizedMetricSpecification': {
                    'MetricName': 'CPUUtilization',
                    'Namespace': 'AWS/RDS',
                    'Dimensions': [
                        {'Name': 'DBClusterIdentifier', 'Value': settings["db_cluster_identifier"]},
                        {'Name': 'Role', 'Value': 'READER'}
                    ],
                    'Statistic': 'Maximum',
                    'Unit': 'Percent'
                },
                'ScaleInCooldown': settings['scale_in_cooldown'],
                'ScaleOutCooldown': settings['scale_out_cooldown']
            }
        )
        logger.info("スケーリングポリシーの設定に成功しました。")
    except Exception as e:
        logger.error(f"スケーリングポリシーの設定に失敗しました。エラー: {e}")

def describe_scalable_targets(service_namespace, resource_id):
    """スケーラブルターゲットの情報を表示"""
    try:
        response = app_scaling_client.describe_scalable_targets(
            ServiceNamespace=service_namespace,
            ResourceIds=[resource_id]
        )
        logger.info("スケーラブルターゲットの情報: %s", response['ScalableTargets'])
    except Exception as e:
        logger.error("スケーラブルターゲット情報の取得に失敗しました。エラー: %s", e)

def describe_scaling_policies(service_namespace, resource_id):
    """スケーリングポリシーの情報を表示"""
    try:
        response = app_scaling_client.describe_scaling_policies(
            ServiceNamespace=service_namespace,
            ResourceId=resource_id
        )
        logger.info("スケーリングポリシーの情報: %s", response['ScalingPolicies'])
    except Exception as e:
        logger.error("スケーリングポリシー情報の取得に失敗しました。エラー: %s", e)

def lambda_handler(event, context):
    # 環境変数のロード
    settings = load_environment_variables()

    # スケーラブルターゲットの登録
    register_scalable_target(settings)

    # 登録情報の表示
    describe_scalable_targets('rds', f'cluster:{settings["db_cluster_identifier"]}')

    # スケーリングポリシーの更新
    update_scaling_policy(settings)

    # スケーリングポリシー情報の表示
    describe_scaling_policies('rds', f'cluster:{settings["db_cluster_identifier"]}')

    return {
        'statusCode': 200,
        'body': 'Success'
    }

5. まとめ

これをLambdaに設定して手動で実行すればOKです。
閾値を変えたい場合は環境変数をLambdaのコンソールから変更して再実行すれば反映されるはずです。
一応loggerの設定は入れていますが、最低限なのでこれ以上入れたい場合は処理を追加すればいいかと思います。