Auto Scaling for Python (Boto3)

次のコード例は、AWS SDK for Python (Boto3)での Auto Scaling を使用してアクションを実行する方法を示しています。

「アクション」は、個々のサービス関数の呼び出し方法を示すコードの抜粋です。

「シナリオ」は、同じサービス内で複数の関数を呼び出して、特定のタスクを実行する方法を示すコード例です。

それぞれの例にはGitHub、へのリンクがあり、コンテキストでコードを設定および実行する方法についての説明が記載されています。

アクション

グループを作成する

次のコード例は、Auto Scaling グループを作成する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def create_group(
            self, group_name, group_zones, launch_template_name, min_size, max_size):
        """
        Creates an Auto Scaling group.

        :param group_name: The name to give to the group.
        :param group_zones: The Availability Zones in which instances can be created.
        :param launch_template_name: The name of an existing Amazon EC2 launch template.
                                     The launch template specifies the configuration of
                                     instances that are created by auto scaling activities.
        :param min_size: The minimum number of active instances in the group.
        :param max_size: The maximum number of active instances in the group.
        """
        try:
            self.autoscaling_client.create_auto_scaling_group(
                AutoScalingGroupName=group_name,
                AvailabilityZones=group_zones,
                LaunchTemplate={
                    'LaunchTemplateName': launch_template_name, 'Version': '$Default'},
                MinSize=min_size, MaxSize=max_size
            )
        except ClientError as err:
            logger.error(
                "Couldn't create group %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいCreateAutoScalingGroup。

グループを削除します

次のコード例は、Auto Scaling グループを削除する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def delete_group(self, group_name):
        """
        Deletes an Auto Scaling group. All instances must be stopped before the
        group can be deleted.

        :param group_name: The name of the group to delete.
        """
        try:
            self.autoscaling_client.delete_auto_scaling_group(
                AutoScalingGroupName=group_name)
        except ClientError as err:
            logger.error(
                "Couldn't delete group %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDeleteAutoScalingGroup。

グループのメトリック収集を無効にする

次のコード例は、Auto ScalingCloudWatch グループに対するメトリクス収集を無効にする方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def disable_metrics(self, group_name):
        """
        Stops CloudWatch metric collection for the Auto Scaling group.

        :param group_name: The name of the group.
        """
        try:
            self.autoscaling_client.disable_metrics_collection(
                AutoScalingGroupName=group_name)
        except ClientError as err:
            logger.error(
                "Couldn't disable metrics %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDisableMetricsCollection。

グループのメトリック収集を有効にする

次のコード例は、Auto ScalingCloudWatch グループに対するメトリクス収集を有効にする方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def enable_metrics(self, group_name, metrics):
        """
        Enables CloudWatch metric collection for Amazon EC2 Auto Scaling activities.

        :param group_name: The name of the group to enable.
        :param metrics: A list of metrics to collect.
        """
        try:
            self.autoscaling_client.enable_metrics_collection(
                AutoScalingGroupName=group_name, Metrics=metrics, Granularity='1Minute')
        except ClientError as err:
            logger.error(
                "Couldn't enable metrics on %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいEnableMetricsCollection。

グループに関する情報を取得する

次のコード例は、Auto Scaling グループに関する情報を取得する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def describe_group(self, group_name):
        """
        Gets information about an Auto Scaling group.

        :param group_name: The name of the group to look up.
        :return: Information about the group, if found.
        """
        try:
            response = self.autoscaling_client.describe_auto_scaling_groups(
                AutoScalingGroupNames=[group_name])
        except ClientError as err:
            logger.error(
                "Couldn't describe group %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            groups = response.get('AutoScalingGroups', [])
            return groups[0] if len(groups) > 0 else None

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeAutoScalingGroups。

インスタンスに関する情報を取得

次のコード例は、Auto Scaling DB インスタンスの情報を取得する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def describe_instances(self, instance_ids):
        """
        Gets information about instances.

        :param instance_ids: A list of instance IDs to look up.
        :return: Information about instances, or an empty list if none are found.
        """
        try:
            response = self.autoscaling_client.describe_auto_scaling_instances(
                InstanceIds=instance_ids)
        except ClientError as err:
            logger.error(
                "Couldn't describe instances %s. Here's why: %s: %s", instance_ids,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return response['AutoScalingInstances']

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeAutoScalingInstances。

スケーリングアクティビティに関する情報を取得する

次のコード例は、Auto Scaling アクティビティに関する情報を取得する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def describe_scaling_activities(self, group_name):
        """
        Gets information about scaling activities for the group. Scaling activities
        are things like instances stopping or starting in response to user requests
        or capacity changes.

        :param group_name: The name of the group to look up.
        :return: The list of scaling activities for the group, ordered with the most
                 recent activity first.
        """
        try:
            response = self.autoscaling_client.describe_scaling_activities(
                AutoScalingGroupName=group_name)
        except ClientError as err:
            logger.error(
                "Couldn't describe scaling activities %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return response['Activities']

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeScalingActivities。

グループの希望容量を設定します

次のコード例は、Auto Scaling グループに必要な容量を設定する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def set_desired_capacity(self, group_name, capacity):
        """
        Sets the desired capacity of the group. Amazon EC2 Auto Scaling tries to keep the
        number of running instances equal to the desired capacity.

        :param group_name: The name of the group to update.
        :param capacity: The desired number of running instances.
        """
        try:
            self.autoscaling_client.set_desired_capacity(
                AutoScalingGroupName=group_name, DesiredCapacity=capacity, HonorCooldown=False)
        except ClientError as err:
            logger.error(
                "Couldn't set desired capacity %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいSetDesiredCapacity。

グループ内のインスタンスの終了

次のコード例は、Auto Scaling グループ内のインスタンスを終了する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def terminate_instance(self, instance_id, decrease_capacity):
        """
        Stops an instance.

        :param instance_id: The ID of the instance to stop.
        :param decrease_capacity: Specifies whether to decrease the desired capacity
                                  of the group. When passing True for this parameter,
                                  you can stop an instance without having a replacement
                                  instance start when the desired capacity threshold is
                                  crossed.
        :return: The scaling activity that occurs in response to this action.
        """
        try:
            response = self.autoscaling_client.terminate_instance_in_auto_scaling_group(
                InstanceId=instance_id, ShouldDecrementDesiredCapacity=decrease_capacity)
        except ClientError as err:
            logger.error(
                "Couldn't terminate instance %s. Here's why: %s: %s", instance_id,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return response['Activity']

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいTerminateInstanceInAutoScalingGroup。

グループの更新

次のコード例は、Auto Scaling グループの設定を更新する方法を示しています。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class AutoScalingWrapper:
    """Encapsulates Amazon EC2 Auto Scaling actions."""
    def __init__(self, autoscaling_client):
        """
        :param autoscaling_client: A Boto3 Amazon EC2 Auto Scaling client.
        """
        self.autoscaling_client = autoscaling_client

    def update_group(self, group_name, **kwargs):
        """
        Updates an Auto Scaling group.

        :param group_name: The name of the group to update.
        :param kwargs: Keyword arguments to pass through to the service.
        """
        try:
            self.autoscaling_client.update_auto_scaling_group(
                AutoScalingGroupName=group_name, **kwargs)
        except ClientError as err:
            logger.error(
                "Couldn't update group %s. Here's why: %s: %s", group_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

• API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいUpdateAutoScalingGroup。

シナリオ

グループとインスタンスの管理

次のコード例は、以下の操作方法を示しています。

• 起動テンプレートとアベイラビリティーゾーンを使用して、、、、、、、、実行中のインスタンスに関する情報を取得します。

• AmazonCloudWatch メトリクスの収集を有効にします。

• グループの希望容量を更新し、インスタンスが起動するのを待ちます。

• グループ内のインスタンスを削除します。

• ユーザーのリクエストや容量の変更に応じて発生するスケーリングアクティビティを一覧表示します。

• CloudWatchメトリクスの統計を取得して、リソースをクリーンアップする。

SDK for Python (Boto3)

注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

コマンドプロンプトからインタラクティブのシナリオを実行します。

def run_scenario(as_wrapper, svc_helper):
    logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')

    print('-'*88)
    print("Welcome to the Amazon EC2 Auto Scaling demo for managing groups and instances.")
    print('-'*88)

    print("This example requires a launch template that specifies how to create\n"
          "EC2 instances. You can use an existing template or create a new one.")
    template_name = q.ask(
        "Enter the name of an existing launch template or press Enter to create a new one: ")
    template = None
    if template_name:
        template = svc_helper.get_template(template_name)
    if template is None:
        inst_type = 't1.micro'
        ami_id = 'ami-0ca285d4c2cda3300'
        print("Let's create a launch template with the following specifications:")
        print(f"\tInstanceType: {inst_type}")
        print(f"\tAMI ID: {ami_id}")
        template_name = q.ask("Enter a name for the template: ", q.non_empty)
        template = svc_helper.create_template(template_name, inst_type, ami_id)
    print('-'*88)

    print("Let's create an Auto Scaling group.")
    group_name = q.ask("Enter a name for the group: ", q.non_empty)
    zones = svc_helper.get_availability_zones()
    print("EC2 instances can be created in the following Availability Zones:")
    for index, zone in enumerate(zones):
        print(f"\t{index+1}. {zone}")
    print(f"\t{len(zones)+1}. All zones")
    zone_sel = q.ask("Which zone do you want to use? ", q.is_int, q.in_range(1, len(zones)+1))
    group_zones = [zones[zone_sel-1]] if zone_sel <= len(zones) else zones
    print(f"Creating group {group_name}...")
    as_wrapper.create_group(group_name, group_zones, template_name, 1, 1)
    wait(10)
    group = as_wrapper.describe_group(group_name)
    print("Created group:")
    pp(group)
    print("Waiting for instance to start...")
    wait_for_instances(group_name, as_wrapper)
    print('-'*88)

    use_metrics = q.ask(
        "Do you want to collect metrics about Amazon EC2 Auto Scaling during this demo (y/n)? ",
        q.is_yesno)
    if use_metrics:
        as_wrapper.enable_metrics(
            group_name, [
                'GroupMinSize', 'GroupMaxSize', 'GroupDesiredCapacity',
                'GroupInServiceInstances', 'GroupTotalInstances'])
        print(f"Metrics enabled for {group_name}.")
    print('-'*88)

    print(f"Let's update the maximum number of instances in {group_name} from 1 to 3.")
    q.ask("Press Enter when you're ready.")
    as_wrapper.update_group(group_name, MaxSize=3)
    group = as_wrapper.describe_group(group_name)
    print("The group still has one running instance, but can have up to three:")
    print_simplified_group(group)
    print('-'*88)

    print(f"Let's update the desired capacity of {group_name} from 1 to 2.")
    q.ask("Press Enter when you're ready.")
    as_wrapper.set_desired_capacity(group_name, 2)
    wait(10)
    group = as_wrapper.describe_group(group_name)
    print("Here's the current state of the group:")
    print_simplified_group(group)
    print('-'*88)
    print("Waiting for the new instance to start...")
    instance_ids = wait_for_instances(group_name, as_wrapper)
    print('-'*88)

    print(f"Let's terminate one of the instances in {group_name}.")
    print("Because the desired capacity is 2, another instance will start.")
    print("The currently running instances are:")
    for index, inst_id in enumerate(instance_ids):
        print(f"\t{index+1}. {inst_id}")
    inst_sel = q.ask(
        "Which instance do you want to stop? ", q.is_int, q.in_range(1, len(instance_ids)+1))
    print(f"Stopping {instance_ids[inst_sel-1]}...")
    as_wrapper.terminate_instance(instance_ids[inst_sel-1], False)
    wait(10)
    group = as_wrapper.describe_group(group_name)
    print(f"Here's the state of {group_name}:")
    print_simplified_group(group)
    print("Waiting for the scaling activities to complete...")
    wait_for_instances(group_name, as_wrapper)
    print('-'*88)

    print(f"Let's get a report of scaling activities for {group_name}.")
    q.ask("Press Enter when you're ready.")
    activities = as_wrapper.describe_scaling_activities(group_name)
    print(f"Found {len(activities)} activities.\n"
          f"Activities are ordered with the most recent one first:")
    for act in activities:
        pp(act)
    print('-'*88)

    if use_metrics:
        print("Let's look at CloudWatch metrics.")
        metric_namespace = 'AWS/AutoScaling'
        metric_dimensions = [{'Name': 'AutoScalingGroupName', 'Value': group_name}]
        print(f"The following metrics are enabled for {group_name}:")
        done = False
        while not done:
            metrics = svc_helper.get_metrics(metric_namespace, metric_dimensions)
            for index, metric in enumerate(metrics):
                print(f"\t{index+1}. {metric.name}")
            print(f"\t{len(metrics)+1}. None")
            metric_sel = q.ask(
                "Which metric do you want to see? ", q.is_int, q.in_range(1, len(metrics)+1))
            if metric_sel < len(metrics)+1:
                span = 5
                metric = metrics[metric_sel - 1]
                print(f"Over the last {span} minutes, {metric.name} recorded:")
                # CloudWatch metric times are in the UTC+0 time zone.
                now = datetime.now(timezone.utc)
                metric_data = svc_helper.get_metric_statistics(
                    metric_dimensions, metric, now-timedelta(minutes=span), now)
                pp(metric_data)
                if not q.ask("Do you want to see another metric (y/n)? ", q.is_yesno):
                    done = True
            else:
                done = True

    print(f"Let's clean up.")
    q.ask("Press Enter when you're ready.")
    if use_metrics:
        print(f"Stopping metrics collection for {group_name}.")
        as_wrapper.disable_metrics(group_name)

    print("You must terminate all instances in the group before you can delete the group.")
    print("Set minimum size to 0.")
    as_wrapper.update_group(group_name, MinSize=0)
    group = as_wrapper.describe_group(group_name)
    instance_ids = [inst['InstanceId'] for inst in group['Instances']]
    for inst_id in instance_ids:
        print(f"Stopping {inst_id}.")
        as_wrapper.terminate_instance(inst_id, True)
    print("Waiting for instances to stop...")
    wait_for_instances(group_name, as_wrapper)
    print(f"Deleting {group_name}.")
    as_wrapper.delete_group(group_name)
    print('-'*88)

    if template is not None:
        if q.ask(f"Do you want to delete launch template {template_name} used in this demo (y/n)? "):
            svc_helper.delete_template(template_name)
            print("Template deleted.")

    print("\nThanks for watching!")
    print('-'*88)


if __name__ == '__main__':
    try:
        wrapper = AutoScalingWrapper(boto3.client('autoscaling'))
        helper = ServiceHelper(boto3.client('ec2'), boto3.resource('cloudwatch'))
        run_scenario(wrapper, helper)
    except Exception:
        logging.exception("Something went wrong with the demo!")

起動テンプレートとメトリックを管理するためにシナリオによって呼び出される関数を定義します。これらの関数は Amazon EC2CloudWatch とアクションをラップします。

class ServiceHelper:
    """Encapsulates Amazon EC2 and CloudWatch actions for the example."""
    def __init__(self, ec2_client, cloudwatch_resource):
        """
        :param ec2_client: A Boto3 Amazon EC2 client.
        :param cloudwatch_resource: A Boto3 CloudWatch resource.
        """
        self.ec2_client = ec2_client
        self.cloudwatch_resource = cloudwatch_resource

    def get_template(self, template_name):
        """
        Gets a launch template. Launch templates specify configuration for instances
        that are launched by Amazon EC2 Auto Scaling.

        :param template_name: The name of the template to look up.
        :return: The template, if it exists.
        """
        try:
            response = self.ec2_client.describe_launch_templates(LaunchTemplateNames=[template_name])
            template = response['LaunchTemplates'][0]
        except ClientError as err:
            if err.response['Error']['Code'] == 'InvalidLaunchTemplateName.NotFoundException':
                logger.warning("Launch template %s does not exist.", template_name)
            else:
                logger.error(
                    "Couldn't verify launch template %s. Here's why: %s: %s", template_name,
                    err.response['Error']['Code'], err.response['Error']['Message'])
                raise
        else:
            return template

    def create_template(self, template_name, inst_type, ami_id):
        """
        Creates an Amazon EC2 launch template to use with Amazon EC2 Auto Scaling.

        :param template_name: The name to give to the template.
        :param inst_type: The type of the instance, such as t1.micro.
        :param ami_id: The ID of the Amazon Machine Image (AMI) to use when creating
                       an instance.
        :return: Information about the newly created template.
        """
        try:
            response = self.ec2_client.create_launch_template(
                LaunchTemplateName=template_name,
                LaunchTemplateData={
                    'InstanceType': inst_type,
                    'ImageId': ami_id})
            template = response['LaunchTemplate']
        except ClientError as err:
            logger.error(
                "Couldn't create launch template %s. Here's why: %s: %s", template_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return template

    def delete_template(self, template_name):
        """
        Deletes a launch template.

        :param template_name: The name of the template to delete.
        """
        try:
            self.ec2_client.delete_launch_template(LaunchTemplateName=template_name)
        except ClientError as err:
            logger.error(
                "Couldn't delete launch template %s. Here's why: %s: %s", template_name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise

    def get_availability_zones(self):
        """
        Gets a list of Availability Zones in the AWS Region of the Amazon EC2 client.

        :return: The list of Availability Zones for the client Region.
        """
        try:
            response = self.ec2_client.describe_availability_zones()
            zones = [zone['ZoneName'] for zone in response['AvailabilityZones']]
        except ClientError as err:
            logger.error(
                "Couldn't get availability zones. Here's why: %s: %s",
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return zones

    def get_metrics(self, namespace, dimensions):
        """
        Gets a list of CloudWatch metrics filtered by namespace and dimensions.

        :param namespace: The namespace of the metrics to look up.
        :param dimensions: The dimensions of the metrics to look up.
        :return: The list of metrics.
        """
        try:
            metrics = list(self.cloudwatch_resource.metrics.filter(
                Namespace=namespace, Dimensions=dimensions))
        except ClientError as err:
            logger.error(
                "Couldn't get metrics for %s, %s. Here's why: %s: %s", namespace, dimensions,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return metrics

    @staticmethod
    def get_metric_statistics(dimensions, metric, start, end):
        """
        Gets statistics for a CloudWatch metric within a specified time span.

        :param dimensions: The dimensions of the metric.
        :param metric: The metric to look up.
        :param start: The start of the time span for retrieved metrics.
        :param end: The end of the time span for retrieved metrics.
        :return: The list of data points found for the specified metric.
        """
        try:
            response = metric.get_statistics(
                Dimensions=dimensions, StartTime=start, EndTime=end, Period=60,
                Statistics=['Sum'])
            data = response['Datapoints']
        except ClientError as err:
            logger.error(
                "Couldn't get statistics for metric %s. Here's why: %s: %s", metric.name,
                err.response['Error']['Code'], err.response['Error']['Message'])
            raise
        else:
            return data


def print_simplified_group(group):
    """
    Prints a subset of data for an Auto Scaling group.
    """
    print(group['AutoScalingGroupName'])
    print(f"\tLaunch template: {group['LaunchTemplate']['LaunchTemplateName']}")
    print(f"\tMin: {group['MinSize']}, Max: {group['MaxSize']}, Desired: {group['DesiredCapacity']}")
    if group['Instances']:
        print(f"\tInstances:")
        for inst in group['Instances']:
            print(f"\t\t{inst['InstanceId']}: {inst['LifecycleState']}")


def wait_for_instances(group_name, as_wrapper):
    """
    Waits for instances to start or stop in an Auto Scaling group.
    Prints the data for each instance after scaling activities are complete.
    """
    ready = False
    instance_ids = []
    instances = []
    while not ready:
        group = as_wrapper.describe_group(group_name)
        instance_ids = [i['InstanceId'] for i in group['Instances']]
        instances = as_wrapper.describe_instances(instance_ids) if instance_ids else []
        if all([x['LifecycleState'] in ['Terminated', 'InService'] for x in instances]):
            ready = True
        else:
            wait(10)
    if instances:
        print(f"Here are the details of the instance{'s' if len(instances) > 1 else ''}:")
        for instance in instances:
            pp(instance)
    return instance_ids

• API の詳細については、「AWS SDK for Python (Boto3) API リファレンス」の以下のトピックを参照してください。

  • CreateAutoScalingGroup

  • DeleteAutoScalingGroup

  • DescribeAutoScalingGroups

  • DescribeAutoScalingInstances

  • DescribeScalingActivities

  • DisableMetricsCollection

  • EnableMetricsCollection

  • SetDesiredCapacity

  • TerminateInstanceInAutoScalingGroup

  • UpdateAutoScalingGroup