ApsaraDB RDS:サーバーレスインスタンスの RCU 構成をスケジュール設定する

準備

1. Python のインストールおよびAlibaba Cloud SDK の依存関係のインストールを行います。

2. APScheduler をインストールします：

   pip install apscheduler

3. ModifyDBInstanceSpec の SDK サンプルプロジェクトをダウンロードします：

   1. ModifyDBInstanceSpec のページを OpenAPI Explorer で開きます。

   2. パラメーター タブで、以下の値を設定します：

      パラメーター	値
      DBInstanceId	rm-bp1t8v93k6e15****
      PayType	Serverless
      Direction	Serverless
      MaxCapacity	8.0
      MinCapacity	0.5
      AutoPause	false
      SwitchForce	true

   3. SDK サンプルコード タブで、言語 ドロップダウンから Python を選択し、プロジェクトのダウンロード をクリックします。注： SDK V2.0 のサンプルプロジェクトをダウンロードしてください。

   4. パッケージを解凍し、_alibabacloud_sample_ ディレクトリを開きます。

定期タスクの設定

APScheduler の定期タスクの内容を sample.py ファイルに追加し、cron トリガーで実行時刻を設定します。cron トリガー hour=7, minute=50 は毎日 07:50 に実行され、hour=9, minute=0 は毎日 09:00 に実行されます。

# -*- coding: utf-8 -*-
# このファイルは自動生成されたものです。編集しないでください。ご協力ありがとうございます。
import os
import sys
from typing import List, Dict
from alibabacloud_rds20140815.client import Client as Rds20140815Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_rds20140815 import models as rds_20140815_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_console.client import Client as ConsoleClient
from alibabacloud_tea_util.client import Client as UtilClient
from apscheduler.schedulers.blocking import BlockingScheduler

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Rds20140815Client:
        """
        AccessKey ID および AccessKey Secret を使用してクライアントを初期化します。
        @return: クライアント
        @throws Exception
        """
        # プロジェクトコードが漏洩すると、AccessKey ペアも漏洩し、アカウント内のリソースのセキュリティが脅かされる可能性があります。以下のサンプルコードは参考用です。
        # セキュリティ上の理由から、Security Token Service (STS) が提供する一時的なアクセス認証情報を使用してください。詳細については、https://www.alibabacloud.com/help/ja/sdk/developer-reference/v2-manage-php-access-credentials をご参照ください。
        config = open_api_models.Config(
            # 必須。ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数が設定されていることを確認してください。
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必須。ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # エンドポイントの詳細については、https://api.aliyun.com/product/Rds をご参照ください。
        config.endpoint = f'rds.aliyuncs.com'
        return Rds20140815Client(config)

    # ModifyDBInstanceSpec API を呼び出して、サーバーレス RDS インスタンスの構成を更新します。
    @staticmethod
    def modify_db_instance_spec(config: Dict[str, float]) -> None:
        """
        DB インスタンス仕様の変更 — コアとなるタスク関数
        """
        client = Sample.create_client()
        serverless_configuration = rds_20140815_models.ModifyDBInstanceSpecRequestServerlessConfiguration(
            max_capacity=config['max_capacity'],
            min_capacity=config['min_capacity'],
            auto_pause=config['auto_pause'],
            switch_force=config['switch_force']
        )
        modify_dbinstance_spec_request = rds_20140815_models.ModifyDBInstanceSpecRequest(
            dbinstance_id='rm-bp1t8v93k6e15****',
            direction='Serverless',
            pay_type='Serverless',
            serverless_configuration=serverless_configuration
        )
        runtime = util_models.RuntimeOptions(
            read_timeout=50000,
            connect_timeout=50000
        )
        try:
            resp = client.modify_dbinstance_spec_with_options(modify_dbinstance_spec_request, runtime)
            ConsoleClient.log(UtilClient.to_jsonstring(resp))
        except Exception as error:
            # 実際のビジネスシーンでは、例外を慎重に処理してください。プロジェクト内で例外を無視しないでください。本例では、エラーメッセージを画面に出力しています。
            print(error.message)
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    def main(args: List[str]) -> None:
        # スケジューラを初期化します。
        scheduler = BlockingScheduler()

        # 08:00～09:00 のピーク時の RCU 範囲：min=4、max=8。
        # ホスト間スケールアップが完了するよう、ピーク開始の 10 分前に 07:50 にスケジュール設定します。
        config_8am = {
            'max_capacity': 8,
            'min_capacity': 4,
            'auto_pause': False,
            'switch_force': True
        }

        # ピーク終了後のデフォルト RCU 範囲：min=0.5、max=8。
        config_9am = {
            'max_capacity': 8,
            'min_capacity': 0.5,
            'auto_pause': False,
            'switch_force': True
        }

        # cron トリガーを使用して定期タスクを作成します。
        scheduler.add_job(Sample.modify_db_instance_spec, 'cron', hour=7, minute=50, args=[config_8am])
        scheduler.add_job(Sample.modify_db_instance_spec, 'cron', hour=9, minute=0, args=[config_9am])

        try:
            scheduler.start()
        except (KeyboardInterrupt, SystemExit):
            pass

if __name__ == '__main__':
    Sample.main(sys.argv[1:])

rm-bp1t8v93k6e15**** をご利用の実際のインスタンス ID に置き換えてください。hour および minute の値は、scheduler.add_job 内の 24 時間表記に従います。ご利用のピーク期間に合わせて調整してください。