なぜそんなことがしたいのか

開発中にいろんなテストで高いインスタンスを立ち上げたりすることがあると思いますが、たまに、稀に、うっかり、落とし忘れて帰ることがあるかもしれません。
まぁ次の日に気づけばダメージは少ないでしょう。ところが連休前だったりとか、しばらく気づかなかったりしたらどうでしょう。そこそこの金額が積みあがって、場合によっては始末書を書くハメにならないとは言い切れません。

AWSには、請求額でアラートを投げる仕組みがありますが、それは既に「ヤバいところまで積みあがってしまった」ことが分かるだけで、予防にはなりません。1ヶ月の予算を3日で使ってしまったとして、後どうするねんていう話ですよ。

そもそも帰る前に確認しろよって話ですが、それをルール化したりチェックリスト作ったりして100%守れますかって話ですよ。確認したリージョンが違ってたとかあるあるじゃないですか。
プログラマーという生き物は、自分がやりたくないことを機械にやらせる為にプログラムを書くものでしょう？
ええ、じゃあやってやりましょう。

AWSのことはAWSで

awscliでインスタンスの状態がとれることは分かっています。
じゃあこれをチョメチョメするのはどこでやるのか？社内のサーバーでcronしてもいいけど、どうせならAWSでやればいいじゃない、1日1,2回ならLambdaでいいじゃない。

Lambdaでawscliするのだ

~~Lambdaでawscliする為のノウハウは、Lambda上でAWSCLIを動かしてS3 Syncするを参考にします。~~

awscliを使ってた時の手順

とりあえずpython3の環境はあるものとしましょう。

> mkdir check-aws
> cd check-aws
> pip install awscli -t .

awsファイルはそのまんま

#!/usr/bin/env python3

import sys
import awscli.clidriver
def main():
    return awscli.clidriver.main()
if __name__ == '__main__':
    result = main()
    sys.exit(result)

xつけてお試し

> chmot +x aws
> ./aws
<略>

次にlambda_function.pyに、ec2とrdsでlarge以上の稼働インスタンスを見つけたらSlackにブン投げる処理を書いていきます。

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

import subprocess
import json
import urllib.request

region_name = {
    'ap-northeast-1': '東京',
    'ap-northeast-2': 'ソウル',
    'ap-southeast-1': 'シンガポール'
}

check_result = []

def check_ec2(region):
    cmd = []
    cmd.append("./aws")
    cmd.append("ec2")
    cmd.append("describe-instances")
    cmd.append("--filter")
    cmd.append("Name=instance-state-name,Values=running")
    cmd.append("--region")
    cmd.append(region)

    result = subprocess.run(cmd, stdout = subprocess.PIPE)
    resjson = json.loads(result.stdout.decode('utf-8'))

    for resv in resjson['Reservations']:
        for ins in resv['Instances']:
            typ = ins['InstanceType'].split('.')[1]
            # 小さいのは許す
            if typ in ['nano', 'micro', 'small', 'medium']:
                continue
            insName = '無名'
            for tag in ins['Tags']:
                if tag['Key'] == 'Name':
                    insName = tag['Value']
            insTyp = ins['InstanceType']
            insLnc = ins['LaunchTime']
            check_result.append('ec2 ' + region_name[region] + ' ' + insName + '(' + insTyp + ') ' + insLnc)

def check_rds(region):
    cmd = []
    cmd.append("./aws")
    cmd.append("rds")
    cmd.append("describe-db-instances")
    cmd.append("--region")
    cmd.append(region)

    result = subprocess.run(cmd, stdout = subprocess.PIPE)
    resjson = json.loads(result.stdout.decode('utf-8'))

    for ins in resjson['DBInstances']:
        typ = ins['DBInstanceClass'].split('.')[2]
        if typ in ['nano', 'micro', 'small', 'medium']:
            continue
        if ins['DBInstanceStatus'] != 'available':
            continue
        insName = ins['DBInstanceIdentifier']
        insTyp = ins['DBInstanceClass']
        check_result.append('rds ' + region_name[region] + ' ' + insName + '(' + insTyp + ')')

def lambda_handler(event, context):
    check_ec2('ap-southeast-1') #シンガポール
    check_ec2('ap-northeast-1') #東京
    check_rds('ap-northeast-1') #東京

    if len(check_result) > 0:
        message = '\nlarge以上のインスタンス稼働状況 <@hogehoge> \n'
        for str in check_result:
            message += str
            message += '\n'
        print(message)
        url = 'https://hooks.slack.com/services/xxxx/yyyy/zzzzzzzz'
        method = 'POST'
        headers = {'Content-Type' : 'application/json'}
        payload = {'text' : message}
        json_data = json.dumps(payload).encode('utf-8')
        request = urllib.request.Request(url, data=json_data, method=method, headers=headers)
        with urllib.request.urlopen(request) as res:
            body = res.read()

if __name__ == '__main__':
    lambda_handler('','')

手元で動くか確認しましょう。行けたらデプロイです。

>zip -r check-aws *

zipファイルでアップロードして、さぁテスト

boto3というライブラリを使う

当初awscliを使ったこの記事を書いたちょっと後に「こんな面倒くさい状態をAWSが放置してるわけない」と調べたらすぐに出てきた。
awscliを無理やりバンドルしなくても、boto3というライブラリでだいたいのAWS操作ができるよう。

では、ec2とrdsでlarge以上の稼働インスタンスを見つけたらSlackにブン投げる処理を書いていきます。

# -*- coding: utf-8 -*-
import json
import urllib.request
import boto3

region_name = {
    'ap-northeast-1': '東京',
    'ap-northeast-2': 'ソウル',
    'ap-southeast-1': 'シンガポール'
}

# 許可するインスタンスIDまたはDB識別子
whitelist = [
    'i-xxxxx',  #アレのやつ
    'i-yyyyy',  #しばらく稼働のソレ
    ]

check_result = []

def check_ec2(region):
    resjson = boto3.client('ec2', region).describe_instances()
    for resv in resjson['Reservations']:
        for ins in resv['Instances']:
            if ins['State']['Name'] != 'running':
                continue
            typ = ins['InstanceType'].split('.')[1]
            if typ in ['nano', 'micro', 'small', 'medium']:
                continue
            if ins['InstanceId'] in whitelist:
                continue
            insName = '無名'
            if 'Tags' in ins:
                for tag in ins['Tags']:
                    if tag['Key'] == 'Name':
                        insName = tag['Value']
            insTyp = ins['InstanceType']
            insLnc = ins['LaunchTime']
            check_result.append('ec2 ' + region_name[region] + ' ' + insName + '(' + insTyp + ') ' + str(insLnc))

def check_rds(region):
    resjson = boto3.client('rds', region).describe_db_instances()
    for ins in resjson['DBInstances']:
        #print(ins)
        typ = ins['DBInstanceClass'].split('.')[2]
        if typ in ['nano', 'micro', 'small', 'medium']:
            continue
        if ins['DBInstanceStatus'] != 'available':
            continue
        if ins['DBInstanceIdentifier'] in whitelist:
            continue
        insName = ins['DBInstanceIdentifier']
        insTyp = ins['DBInstanceClass']
        check_result.append('rds ' + region_name[region] + ' ' + insName + '(' + insTyp + ')')

def lambda_handler(event, context):
    check_ec2('ap-northeast-2') #ソウル
    check_ec2('ap-southeast-1') #シンガポール
    check_ec2('ap-northeast-1') #東京
    check_rds('ap-southeast-1') #シンガポール

    print(check_result)

    if len(check_result) > 0:
        message = "\nlarge以上のインスタンス稼働状況 <@hogehoge> <@hogahoga> \n"
        for str in check_result:
            message += str
            message += '\n'
        print(message)
        url = 'https://hooks.slack.com/services/xxxx/yyyy/zzzzzzzz'
        method = 'POST'
        headers = {'Content-Type' : 'application/json'}
        payload = {'text' : message}
        json_data = json.dumps(payload).encode("utf-8")
        request = urllib.request.Request(url, data=json_data, method=method, headers=headers)
        with urllib.request.urlopen(request) as res:
            body = res.read()

awscliを使ってた時と違うのは、ブラウザから編集・デバッグできるようになったところです。

おわりに

Slack通知メッセージには、状況を分かってインスタンスを落とせる人（通知を無視しない人、というのも案外重要です）複数人にメンション付けておくと良いでしょう。Slack使ってなかったらSNSなり適当にアレンジすれば良いでしょう。
今回はlarge以降のEC2とRDSだけを標的にしましたが、インスタンスの数だとか他のサービスについても同様に対応できるのではないでしょうか。しらんけど。

いくつか注意点があります。
・PowerUserAccessのロールをつけておくこと
・タイムアウトをちょい多めに設定すること
・メモリはちょい多めに設定すること
メモリは128MBでも足りてるっぽいですが、CPUがクソ貧弱なので1024MBぐらいにはしてやった方が良いと思います。要調整。

テストしてうまくいけば（いかなければエラーメッセージ見て何とかしましょう）トリガを設置して完了です。
EventBridgeでスケジュール式にcron(0 9 ? * MON-FRI *)とかやっとけばいいんじゃないですか。

では。

(2023/10/29) awscliじゃなくboto3を使う方法に更新（長らく放置してた・・）