この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
こんにちは、muroです。AWS運用かんたん自動化ツール「opswitch」の開発・運用を担当しています。opswitchはboto3
をがんがん使ってAWSを操作します。今回はboto3
のモックライブラリ「moto」を利用して、EC2インスタンスの起動をテストしてみます。ただし、EC2インスタンスを起動するだけではなく、よくあるトラブルに対応した実装を試します。
前回記事はこちら。
シナリオ
EC2インスタンス起動時にInsufficientInstanceCapacity
エラーが発生し、インスタンスを起動できないことがあります。
EC2 インスタンスの開始または起動時に発生する、InsufficientInstanceCapacity エラーのトラブルシューティング - AWS
数回リトライしても起動できなかった場合、別のインスタンスタイプに変更して起動させるようにしてみます。
環境
今回のテストに使用した環境は以下の通りです。
環境 | バージョン |
---|---|
Python | 3.9.13 |
boto3 | 1.23.3 |
moto | 3.1.9 |
pytest | 7.1.2 |
pytest-mock | 3.7.0 |
retrying | 1.3.3 |
またテストプロジェクトの構成は以下の通りです。
.
├── main.py
└── test_main.py
やってみる
テスト対象コード
main.py
import boto3
from retrying import retry
from botocore.exceptions import ClientError
import logging
logger = logging.getLogger()
def start_instances(instance_ids):
"""
EC2インスタンスを起動します。
"""
ec2 = boto3.client('ec2')
ec2.start_instances(
InstanceIds=instance_ids
)
def change_instance_type(instance_id, alt_instance_type):
"""
EC2インスタンスのインスタンスタイプを変更します。
"""
ec2 = boto3.client('ec2')
ec2.modify_instance_attribute(
InstanceId=instance_id,
Attribute='instanceType',
Value=alt_instance_type
)
@retry(wait_exponential_multiplier=1000, stop_max_attempt_number=5)
def start_instance(instance_id):
"""
EC2インスタンスの起動をリトライするラッパー関数です。
"""
logger.info('try start_instances...')
start_instances([instance_id])
def force_start_instance(instance_id, alt_instance_type):
"""
EC2インスタンスを起動します。
InsufficientInstanceCapacityエラー時はインスタンスタイプを変更して
再度EC2インスタンスの起動を試みます。
"""
try:
start_instance(instance_id)
except ClientError as ce:
logger.error(ce)
error_code = ce.response['Error']['Code']
if error_code == 'InsufficientInstanceCapacity':
change_instance_type(instance_id, alt_instance_type)
start_instance(instance_id)
retryingについて
関数に@retry
デコレータを指定するだけでその関数をリトライしてくれる便利なライブラリです。今回はエクスポネンシャル・バックオフで時間をおいて最大5回リトライするように指定しました。retryingの詳しい使い方は以下の記事を参考にしてください。
エラーハンドリングについて
botocore.exceptions.ClientError
をキャッチして、エラーコードがInsufficientInstanceCapacity
エラーかどうかを判定しています。インスタンスタイプを変更して、さらにstart_instance()
を呼び出しますが、それでも起動できなかった場合は例外が発生して終了します。boto3
のエラーハンドリングについては以下のリファレンスも参考にしてください。
Error handling — Boto3 Docs 1.24.61 documentation
テストコード
今回もmoto
を利用してEC2インスタンスをモックしています。さらに、pytest-mock
を利用して、start_instances
関数をモックしています。インスタンスタイプが代替指定のインスタンスタイプでない場合InsufficientInstanceCapacity
エラーを発生させます。インスタンスタイプが代替指定のインスタンスタイプの場合にEC2インスタンスを起動して、結果をassert
します。
test_main.py
import boto3
from moto import mock_ec2
import main
from botocore.exceptions import ClientError
import logging
logger = logging.getLogger()
@mock_ec2
def test_force_start_instances(mocker):
expected_number_of_instances = 1
initial_instance_type = 't2.medium'
input_alt_instance_type = 't3.medium'
input_instance_id = run_instances(
expected_number_of_instances,
initial_instance_type
)[0]
# 例外を発生させるようにMockを設定します。
def mock_start_instances(instance_ids):
instance = describe_instance(instance_ids[0])
instance_type = instance['InstanceType']
if instance_type == input_alt_instance_type:
ec2 = boto3.client('ec2')
ec2.start_instances(
InstanceIds=instance_ids
)
else:
raise ClientError(
{'Error': {'Code': 'InsufficientInstanceCapacity'}},
'StartInstances'
)
mocker.patch('main.start_instances', side_effect=mock_start_instances)
# テストを実行します。
main.force_start_instance(input_instance_id, input_alt_instance_type)
# インスタンスが起動されたかどうかテストします。
instance = describe_instance(input_instance_id)
assert(instance['InstanceType'] == input_alt_instance_type)
assert(instance['State']['Name'] == 'running')
def run_instances(expected_number_of_instances, instance_type):
# 期待するインスタンス数の分だけインスタンスを起動します。
ec2 = boto3.client('ec2')
images = ec2.describe_images()['Images']
response = ec2.run_instances(
ImageId=images[0]['ImageId'],
MinCount=expected_number_of_instances,
MaxCount=expected_number_of_instances,
InstanceType=instance_type
)
# インスタンスIDのリストを取得します。
instance_ids = [x['InstanceId'] for x in response['Instances']]
# インスタンスを停止します。
ec2.stop_instances(
InstanceIds=instance_ids
)
# インスタンスIDのリストを返します。
return instance_ids
def describe_instance(instance_id):
ec2 = boto3.client('ec2')
response = ec2.describe_instances(InstanceIds=[instance_id])
return response['Reservations'][0]['Instances'][0]
テスト実行
それではテストを実行してみます。今回のテスト対象コードはlogger
を仕込んでいるので、その出力を表示させながら実行します。
python -m pytest --log-cli-level=INFO -s \
--log-cli-format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s"
========================================== test session starts ==========================================
platform linux -- Python 3.9.13, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/ユーザー/プロジェクト
plugins: cov-3.0.0, mock-3.7.0
collected 1 item
test_main.py::test_force_start_instances
--------------------------------------------- live log call ---------------------------------------------
17:57:01.855 INFO Found credentials in environment variables.
17:57:02.041 INFO try start_instances...
17:57:04.072 INFO try start_instances...
17:57:08.139 INFO try start_instances...
17:57:16.193 INFO try start_instances...
17:57:32.242 INFO try start_instances...
17:57:32.299 ERROR An error occurred (InsufficientInstanceCapacity) when calling the StartInstances operation: Unknown
17:57:32.308 INFO try start_instances...
PASSED
========================================== 1 passed in 31.20s ===========================================
2秒・4秒・8秒・16秒とリトライ間隔をあけてリトライされ、エラーの後テストが成功しました。
まとめ
retrying
ライブラリを利用することでエクスポネンシャル・バックオフで待機時間を延ばしながら関数をリトライさせることができました。また、moto
でEC2インスタンスをモックしつつpytest-mock
を利用して複雑なエラー条件をテストすることができました。なかなか発生させにくいエラーとそのハンドリングであっても、ライブラリを利用することで、ある程度再現させられるのはとても便利だと思いました。以上、ご参考になりましたら幸いです。