PythonでAWS EC2を起動・停止する
はじめに
ブラウザを開いてAWSコンソール画面にログインして、EC2を起動・停止するのが面倒くさい。
なので、プログラム(Python)で起動するようにしてみた。
環境
- Python 3.6
- boto3
ソースコード
boto3のドキュメントを参考にしつつ作成
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)
参考にしたサイト
例外の定義はコチラ(ファイル名: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やらロードバランサーやら)を起動・停止するのに使った一部を抜粋