もなかアイスの試食品

「とりあえずやってみたい」そんな気持ちが先走りすぎて挫折が多い私のメモ書きみたいなものです.

PythonでAWS EC2を起動・停止する

はじめに

ブラウザを開いてAWSコンソール画面にログインして、EC2を起動・停止するのが面倒くさい。

なので、プログラム(Python)で起動するようにしてみた。

環境

ソースコード

boto3のドキュメントを参考にしつつ作成

boto3.amazonaws.com

EC2を起動・停止するメイン処理はこんな感じになった

ファイル名は「aws.py」で作成

# -*- coding: utf-8 -*-

import boto3
import json
import logger
from exception import NotSupportError, NotFoundResource
from botocore.exceptions import ClientError

with open('credentials.json', 'r') as stream:
    credentials = json.load(stream)

class EC2:
    """
    EC2を起動・停止する機能を提供する
    """

    def __init__(self, instance_id, region_name):
        """
        コンストラクタ
        :param instance_id: EC2のインスタンスID
        :param region_name: リージョン名
        """
        self.__log = logger.Logger('EC2')
        self.__instance_id = instance_id
        self.__local_session = boto3.session.Session(
            aws_access_key_id=credentials['AccessKey'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            region_name=region_name
        )

    def start(self):
        """
        起動する
        """
        ec2 = self.__create_ec2()
        try:
            state_name = ec2.state['Name']
        except AttributeError as e:
            # AttributeErrorは、昔あったインスタンスIDを使うと、(Pythonの)インスタンスは生成されるが、
            # 属性stateにアクセスできず例外が発生する
            self.__log.error(e)
            raise NotFoundResource(e)

        self.__log.debug('state is %s' % (state_name,))
        if state_name == 'running':
            self.__log.info('ec2(%s) is already started.' % (self.__instance_id,))
            return
        elif state_name in {'pending', 'shutting-down', 'terminated', 'stopping'}:
            raise NotSupportError('ec2(%s) is %s.' % (self.__instance_id, state_name))
        elif state_name == 'stopped':
            self.__log.info('try start ec2(%s).' % (self.__instance_id,))
            ec2.start()
            # runningになるまで待機
            ec2.wait_until_running()
            self.__log.info('ec2(%s) is started.' % (self.__instance_id,))
            pass
        else:
            raise NotSupportError('Unknown status: %s' % (state_name,))

    def stop(self):
        """
        停止する
        """
        ec2 = self.__create_ec2()
        try:
            state_name = ec2.state['Name']
        except AttributeError as e:
            # AttributeErrorは、昔あったインスタンスIDを使うと、(Pythonの)インスタンスは生成されるが、
            # 属性stateにアクセスできず例外が発生する
            self.__log.error(e)
            raise NotFoundResource(e)

        self.__log.debug('state is %s' % (state_name,))
        if state_name == 'stopped':
            self.__log.info('ec2(%s) is already stopped.' % (self.__instance_id,))
            return
        elif state_name in {'pending', 'shutting-down', 'terminated', 'stopping'}:
            raise NotSupportError('ec2(%s) is %s.' % (self.__instance_id, state_name))
        elif state_name == 'running':
            self.__log.info('try stop ec2(%s).' % (self.__instance_id,))
            ec2.stop()
            self.__log.info('ec2(%s) is stopping.' % (self.__instance_id,))
            pass
        else:
            raise NotSupportError('Unknown status: %s' % (state_name,))

    def __create_ec2(self):
        """
        EC2を操作するオブジェクトを生成する
        :return: EC2を操作するオブジェクト
        """
        self.__log.debug('ec2: %s' % (self.__instance_id,))
        ec2_resource = self.__local_session.resource('ec2')
        try:
            return ec2_resource.Instance(self.__instance_id)
        except ClientError as e:
            self.__log.error(e)
            raise NotFoundResource(e)

credentials.jsonは、AWS IAMで作成した認証情報を保存している。

ちなみに、ログを出力しているクラスの定義はコチラ(ファイル名:logger.py)

# -*- coding: utf-8 -*-

from logging import Formatter, handlers, StreamHandler, getLogger, INFO


class Logger:
    def __init__(self, name=__name__):
        self.logger = getLogger(name)
        self.logger.setLevel(INFO)
        formatter = Formatter("[%(asctime)s] [%(process)d] [%(name)s] [%(levelname)s] %(message)s")

        # stdout
        handler = StreamHandler()
        handler.setLevel(INFO)
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def debug(self, msg):
        self.logger.debug(msg)

    def info(self, msg):
        self.logger.info(msg)

    def warn(self, msg):
        self.logger.warning(msg)

    def error(self, msg):
        self.logger.error(msg)

    def critical(self, msg):
        self.logger.critical(msg)

参考にしたサイト

qiita.com

例外の定義はコチラ(ファイル名:exception.py)

# -*- coding: utf-8 -*-


class NotSupportError(Exception):
    """
    サポートしていない動作が行われようとしたときに発生
    """
    pass


class NotFoundResource(Exception):
    """
    リソースが見つからないときに発生
    """
    pass

使い方

例えば、以下のようなJsonファイルがあった場合

{
    "EC2": [
      {
        "InstanceId": "i-xxxxxxxxxxxxxxx",
        "RegionName": "ap-northeast-1"
      }
    ]
}

以下のようにするとEC2が起動出来る

# -*- coding: utf-8 -*-
import aws

# jsonの読み込み処理(省略)


def start_ec2(ec2_machines):
    if ec2_machines is None or len(ec2_machines) == 0:
        return
    for machine in ec2_machines:
        ec2 = aws.EC2(
            instance_id=machine['InstanceId'],
            region_name=machine['RegionName']
        )
        ec2.start()

if __name__ == '__main__':
    start_ec2(json['EC2'])

停止は、EC2クラスのstopメソッドを呼ぶようにする

おわりに

今回のプログラムはサービス全体(EC2やらロードバランサーやら)を起動・停止するのに使った一部を抜粋

Pythonロードバランサーの生成・削除、Route53の登録も出来たので、それらについてはまたいつか・・・