Lambda(python)でEC2スポットリクエストを実行する(EC2スポットインスタンス自動化3)

背景(再掲)

個人向けに最低費用で運用するため、Ec2をスポットインスタンスで使用しているが、時々スポットリクエストが中断される。その都度、再度スポットリクエストで立ち上げなおしているが、この処理を自動化したい。

大まかな流れ(再掲)

スポットインスタンスの中断はイベント通知されるので、それをトリガとして、立ち上げ処理を行う。

全体の流れは第1回参照。本記事はスポットリクエストの中断イベントから対象の情報を取得する部分が対象。

記事の全体

1. AWS BackupでEC2のAMIが作成されるので、その最新のAMI IDをイベントから取得してParameterStoreに登録する。
2. AWS Lambda(python)でLINEへ通知する。
3. AWS Lambda(python)でEC2スポットリクエストを実行する。←今回の内容
4. AWS Lambda(Python,boto3)でElasticIPを付け替える。
5. Step FunctionでLambdaでスポットリクエストを実行し、ElasticIPを付け替える。

調べたこと

①EC2スポットリクエストの中断イベント

スポットリクエストが中断する際には2分前にイベントが発行される。

(サンプル転記).json

{
    "version": "0",
    "id": "12345678-1234-1234-1234-123456789012",
    "detail-type": "EC2 Spot Instance Interruption Warning",
    "source": "aws.ec2",
    "account": "123456789012",
    "time": "yyyy-mm-ddThh:mm:ssZ",
    "region": "us-east-2",
    "resources": ["arn:aws:ec2:us-east-2:123456789012:instance/i-1234567890abcdef0"],
    "detail": {
        "instance-id": "i-1234567890abcdef0",
        "instance-action": "action"
    }
}

ちなみに、EC2内部でもインスタンスメタデータから、対象になったことが分かる模様。定期的にチェックしておくことで、EC2内の終了処理を実行する・データ退避する(1分以内ぐらい?)の処理ができそう。

Amazon EC2 が、スポットインスタンスを停止または終了のためにマークした場合、インスタンスメタデータ内に instance-action 項目が含まれるようになります。

イベントからの情報収集

イベント内のdetailにインスタンスIDが入っているので、parameter storeのキーに含めているホスト名を取得する。

instance_id = event['detail']['instance-id']

(取得部分のお試しプログラム).py

import json

ec2 = boto3.client('ec2', region_name=REGION)
response = ec2.describe_instances(InstanceIds=[インスタンスID])
tags = response['Reservations'][0]['Instances'][0]['Tags']
print('tags for ' + instance_id + ':' + json.dumps(tags))

for idx, keyvalue in enumerate(tags):
    if( keyvalue['Key'] == 'Name' ):
        print('hostname:' + keyvalue['Value'])

②Lambda(Python,boto3)でのスポットリクエスト発行

pythonでスポットリクエストを発行するには、run_instancesでスポットリクエストのパラメータを設定する必要がある。

(run_instancesのオプション).py

    InstanceMarketOptions={
        'MarketType': 'spot',
        'SpotOptions': {
            'MaxPrice': 'string',
            'SpotInstanceType': 'one-time'|'persistent',
            'BlockDurationMinutes': 123,
            'ValidUntil': datetime(2015, 1, 1),
            'InstanceInterruptionBehavior': 'hibernate'|'stop'|'terminate'
        }
    }

request_spot_instancesというAPIもあるが、古くて利用は推奨されていない。

③run_instancesを読み解く。

[指定必須]

• MaxCount (integer)
• MinCount (integer)

[指定したい]

• ImageId (string)
• InstanceType (string)
• SecurityGroupIds (list)
• Placement (dict)
  • AvailabilityZone (string)
• SubnetId (string)
• KeyName (string)
• IamInstanceProfile (dict) ※IAM roleのARNを渡す
• InstanceMarketOptions
  • {"MarketType": "spot","SpotOptions": {"SpotInstanceType": "one-time"}}'
    • MarketType 'spot'で設定
    • SpotOptions
      • SpotInstanceType 'one-time'で設定
• TagSpecifications (list)
  • ResourceType (string)
  • Tags (list)
    • Key (string)
    • Value (string)

[テスト時に使いたい]

• DryRun (boolean)

④ParameterStoreの準備

スポットリクエストで必要なパラメータはParameterStoreに保存しておき、Lambdaから参照するようにします。

/SpotRequest/ホスト名/IamInstanceProfileARN ・・・EC2に設定するIAM InstanceProfileのARN
/SpotRequest/ホスト名/availabilityZones	・・・スポットインスタンスを起動するAZ候補のリスト
/SpotRequest/ホスト名/instanceTypes	・・・起動したいインスタンスタイプ候補のリスト
/SpotRequest/ホスト名/keypair	・・・EC2を起動する時に使うキーペア名
/SpotRequest/ホスト名/securityGroupIds	・・・EC2に設定するセキュリティグループ
/SpotRequest/ホスト名/subnet-id/ap-northeast-1a	・・・AZのサブネットID
/SpotRequest/ホスト名/subnet-id/ap-northeast-1c	・・・AZのサブネットID
/SpotRequest/ホスト名/subnet-id/ap-northeast-1d　・・・AZのサブネットID

Lambda関数の作成

※ログ多めです。

(requestSpotInstance).py

import boto3
import botocore
from botocore.exceptions import ClientError
import os
from datetime import datetime,date
import json
import logging

logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
logger.setLevel(logging.INFO)
logger.addHandler(handler)

# パラメータで指定する
REGION = os.getenv("AWS_REGION")
DRYRUN = False

def get_parameters(param_key):
    ssm = boto3.client('ssm', region_name=REGION)
    response = ssm.get_parameters(Names=[param_key])
    return response['Parameters'][0]['Value']

# datetimeをjson dumpsするための変換関数
def json_serial(obj):
    # 文字列に変換
    if isinstance(obj, (datetime,date)):
        return obj.isoformat()
    raise TypeError ("Type %s not serializable" % type(obj))

def lambda_handler(event, context):
    
    logger.info('----- recieve event -----')
    logger.info('event:' + json.dumps(event))
    instance_id = event['detail']['instance-id']
    
    logger.info('----- get hostname from instance id -----')
    ec2 = boto3.client('ec2', region_name=REGION)
    response = ec2.describe_instances(InstanceIds=[instance_id])
    tags = response['Reservations'][0]['Instances'][0]['Tags']
    print('tags for ' + instance_id + ':' + json.dumps(tags))
    
    for idx, keyvalue in enumerate(tags):   
        if( keyvalue['Key'] == 'Name' ):
            print('hostname:' + keyvalue['Value'])
            hostname = keyvalue['Value']
    
    logger.info('----- get target AZ/InstanceType from parameter store -----')
    logger.info('Target Availability Zones: ' + get_parameters('/SpotRequest/' + hostname + '/availabilityZones'))
    logger.info('Target Instance Types: ' + get_parameters('/SpotRequest/' + hostname + '/instanceTypes'))
    
    logger.info('----- EC2 Launch Parameters -----')
    logger.info('ImageId:' + get_parameters('/EC2Backup/latestAMI/' + hostname ))
    logger.info('SecurityGroupIds:' + get_parameters('/SpotRequest/' + hostname + '/securityGroupIds'))
    logger.info('IamInstanceProfile:' + json.dumps({ "Arn": get_parameters('/SpotRequest/' + hostname + '/IamInstanceProfileARN') }))
    logger.info('KeyName:' + get_parameters('/SpotRequest/' + hostname + '/keypair'))
    logger.info('TagSpecifications:' +
        json.dumps([
            {"ResourceType": "instance","Tags": [
                { "Key": "AWSBackup","Value": "EC2"},
                { "key": "Name","Value": hostname }
            ]}]
        )
    )
    logger.info('InstanceMarketOptions={ MarketType: spot, SpotOptions: { SpotInstanceType: one-time }')
    returnMessage = 'All spot request was rejected.'

    instance_ids  =[]
    for targetInstanceType in get_parameters('/SpotRequest/' + hostname + '/instanceTypes').split(','):
        for targetAZ in get_parameters('/SpotRequest/' + hostname + '/availabilityZones').split(','):
            logger.info('----- Spot Request(' + targetInstanceType + ':' + targetAZ + ') -----')
    
            logger.info('InstanceType:' + targetInstanceType)
            logger.info('Placement: { AvailabilityZone:' + targetAZ +'}')
            logger.info('SubnetId:' + get_parameters('/SpotRequest/' + hostname + '/subnet-id/' + targetAZ ))
    
            runFlag = False
            errorFlag = False
            # Launch Instances
            client = boto3.client('ec2',region_name=REGION )
            try:
                run_response = client.run_instances(
                    MaxCount = 1,
                    MinCount = 1,
                    ImageId = get_parameters('/EC2Backup/latestAMI/' + hostname ),
                    SecurityGroupIds = get_parameters('/SpotRequest/' + hostname + '/securityGroupIds').split(','),
                    InstanceType = targetInstanceType,
                    IamInstanceProfile = { "Arn": get_parameters('/SpotRequest/' + hostname + '/IamInstanceProfileARN') },
                    Placement = { "AvailabilityZone": targetAZ },
                    SubnetId = get_parameters('/SpotRequest/' + hostname + '/subnet-id/' + targetAZ ),
                    KeyName = get_parameters('/SpotRequest/' + hostname + '/keypair'),
                    TagSpecifications = [
                        {
                        "ResourceType": "instance",
                        "Tags": [
                            { "Key": "AWSBackup","Value": "EC2"},
                            { "Key": "Name","Value": hostname }
                            ]
                        }
                    ],
                    InstanceMarketOptions={
                        'MarketType': 'spot',
                        'SpotOptions': {
                            'SpotInstanceType': 'one-time',
                        }
                    },
                    DryRun = DRYRUN
                )
            except ClientError as e:
                if 'DryRunOperation' not in str(e):
                    logger.error('run_instance was failed. detail info: ' + json.dumps(e.response['Error']) )
                    errorFlag = True
                else:
                    #DryRunの場合
                    logger.info( '----- dry run was completed successfully -----')
                    returnMessage = ( 'DryRun was complete. targetInstanceType [' +
                                        targetInstanceType + '] targetAZ [' + targetAZ + ']' )
                    runFlag = True
                    break
    
            # run_instancesが問題なく完了し、DryRunでもない場合
            if errorFlag == False and runFlag == False:
                logger.info('run_instances:' + json.dumps(run_response, default=json_serial))
    
                for instance in run_response['Instances']:
                    instance_ids.append( instance['InstanceId'] )
                response = { "LaunchedInstances" : instance_ids }
                print('LaunchedInstances=['+ json.dumps(response) + ']')
    
                returnMessage = ( 'Spot request was complete. new instance id [ ' +
                                 json.dumps(response) + '] ' + targetInstanceType + 
                                 '[' + targetInstanceType + '] + targetAZ [' + targetAZ + ']' )
                runFlag = True
                break
        
        if runFlag:
            break

    responseData = {
        'instance_id' : instance_ids,
        'notify_message' : 
        [
            {
                'type' : 'text',
                'text' : returnMessage
            },
            {
                'type' : 'sticker', 
                'packageId' : 6370, 
                'stickerId' : 11088016
            }
        ]
    }
    return ( responseData)