AWS EC2インスタンスの自動起動停止の制御手法

概要

AWS EC2インスタンスを利用しています。費用を節約するために、PythonでEC2インスタンスを自動起動停止のモジュールを作成しました。

ソース

以下のスースを作成しました。

import boto3
import os, sys, traceback

ec2_instances = None
dict_instance_names = {}
dict_instance_ids = {}


def to_exception_detail(exception, includes_trace=False):
    exc_type, exc_obj, exc_tb = sys.exc_info()
    fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
    lineno = exc_tb.tb_lineno
    trace = traceback.format_exc() if includes_trace else ''
    return 'at [{0}]({1}:{2}) -> {3}{4}'.format(type(exception).__name__, fname, lineno, exception, trace)


class Ec2InstanceOp(object):
    global ec2_instances
    global dict_instance_names
    global dict_instance_ids
    ec2_instances = boto3.client('ec2')

    instances = ec2_instances.describe_instances()
    for reservations in instances['Reservations']:
        for instance in reservations['Instances']:
            tags = instance['Tags']
            for taginfo in tags:
                if taginfo['Key'] == 'Name':
                    dict_instance_ids[instance['InstanceId']] = taginfo['Value']
                    dict_instance_names[taginfo['Value']] = instance['InstanceId']

    def start(self, ec2ins_names):
        # ec2 = boto3.resource('ec2')
        # instance = ec2.Instance(intance_id)
        try:
            target_ec2ins_names = []
            failed_instances = []
            for ec2ins_name in ec2ins_names:
                if not self.is_openning(ec2ins_name):
                    target_ec2ins_names.append(ec2ins_name)
            ec2ins_ids = self.__get_ec2_instanceids(target_ec2ins_names)

            if ec2ins_ids is None or len(ec2ins_ids) == 0:
                return True, failed_instances

            response = ec2_instances.start_instances(InstanceIds=ec2ins_ids)
            if 'StartingInstances' in response and len(response['StartingInstances']) > 0:
                starting_instances = response['StartingInstances']
                for starting_instance in starting_instances:
                    instanceid = starting_instance['InstanceId']
                    if 'CurrentState' in starting_instance:
                        state = starting_instance['CurrentState']
                        if 'Code' in state:
                            code = int(state['Code'])
                            if code != 0:
                                failed_instances.append(dict_instance_ids[instanceid])
            if len(failed_instances) == 0:
                return True, failed_instances
            else:
                return False, failed_instances
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def stop(self, ec2ins_names):
        try:
            target_ec2ins_names = []
            failed_instances = []
            for ec2ins_name in ec2ins_names:
                if self.is_openning(ec2ins_name):
                    target_ec2ins_names.append(ec2ins_name)
            ec2ins_ids = self.__get_ec2_instanceids(target_ec2ins_names)

            if ec2ins_ids is None or len(ec2ins_ids) == 0:
                return True, failed_instances
            response = ec2_instances.stop_instances(InstanceIds=ec2ins_ids)
            if 'StoppingInstances' in response and len(response['StoppingInstances']) > 0:
                stopping_instances = response['StoppingInstances']
                for stopping_instance in stopping_instances:
                    instanceid = stopping_instance['InstanceId']
                    if 'CurrentState' in stopping_instance:
                        state = stopping_instance['CurrentState']
                        if 'Code' in state:
                            code = int(state['Code'])
                            if code != 64:
                                failed_instances.append(dict_instance_ids[instanceid])
            if len(failed_instances) == 0:
                return True, failed_instances
            else:
                return False, failed_instances
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def is_openning(self, instance_name):
        try:
            state = self.__get_ec2_state(self.get_ec2_instanceid(instance_name))
            if state is not None and int(state) == 16:
                return True
            return False
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def get_ec2_instanceid(self, instance_name):
        try:
            if instance_name in dict_instance_names:
                return dict_instance_names[instance_name]
            return None
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def __get_ec2_instanceids(self, instance_names):
        try:
            return list(map(lambda instance_name: dict_instance_names[
                instance_name] if instance_name in dict_instance_names else None, instance_names))
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def __get_ec2_state(self, instance_id):
        try:
            instanceIds = []
            instanceIds.append(instance_id)
            response = ec2_instances.describe_instances(InstanceIds=instanceIds)
            if 'Reservations' in response and len(response['Reservations']) > 0:
                reservations = response['Reservations'][0]
                if 'Instances' in reservations and len(reservations['Instances']) > 0:
                    instances = reservations['Instances'][0]
                    if 'State' in instances:
                        state = instances['State']
                        if 'Code' in state:
                            return state['Code']

            return None
        except Exception as e:
            raise Exception(to_exception_detail(e))

補足

1. 上記のモジュールを動かすために、awscli 用の環境変数を設定することが必要です。以下のように aws configure コマンドで設定することができます。

# aws configure
AWS Access Key ID [****************ETRQ]: 
AWS Secret Access Key [****************oY/N]:
Default region name [ap-northeast-1]:
Default output format [None]:

2.対象の EC2 インスタンスに対して、Name タグを登録することが必要です。その値は EC2 インスタンスの名前となります。