AWS NAT ゲートウェイ を理解した上で起動＆停止 Lambda を用意する

実現したいこと

既存のAWSのバックエンドシステムにおいて、コスト削減のため開発環境とステージング環境のNATゲートウェイ（NAT Gateway）を夜間と休日の間は停止させるべく、NATゲートウェイを起動・停止させるLambdaを作成します。

待て、まだコードを書こうとするな

「NATゲートウェイ　自動　停止」とかで検索すると少ないながらもいくつか記事が出てきますが、理解せずにそれをまるまるコピペして実行させるだけではうまくいかないことが多いですし、望まないNAT削除をしてしまう可能性があります。
NATゲートウェイ、ルートテーブル、Elastic IPなどが自分の開発しているシステムではどのように設定されていて、どのリソースを操作すれば良いかをちゃんと理解する必要がありました。

こちらは開発していたシステムの構成図を、NATゲートウェイ周りだけを抽出して描いたものです。
手描きでごめんなさい。（手で描いた方が頭に残ると思って。）
これを基にNATゲートウェイ周りを説明します。

NATゲートウェイ

まず注目すべきはNATゲートウェイの位置です。
パブリックサブネットにあります。
NATゲートウェイの役割は、簡単にいうとプライベートパブリックのリソースが外部にインターネットでアクセスできるようにすることです。

また、NATゲートウェイにはElastic IPが割り当てられます。
注意しておきたいのは、このElastic IPは動的に割り当てられるということと、個数に上限があるということです。

そして最後に大事な前提がありまして、それは、NATゲートウェイは「起動」と「停止」ができず、「作成」と「削除」しかできないということです。
タイトル詐欺のようになってしまいますが、これから作るのはNATゲートウェイの 作成＆削除Lambda ということになります。

【余談】上記の画像ではNATゲートウェイは片方のパブリックサブネットにしかありませんが、耐障害性を高めるのであれば両方のパブリックサブネットに置いてマルチAZ構成にするのも良いでしょう。

ルーティング

ただNATゲートウェイを作成したり削除したりするだけなら正直そこまでお勉強は必要ないです。
しかし事態を若干ややこしくするのは、NAT周りのルーティングも作成や削除の度に自分で設定してあげる必要があるということです。
とは言っても一通りやってみて思ったのは、理解してしまえば全然怖くない、ということです。
理解すべきはたった2つ。

1. プライベートサブネットは、NATゲートウェイをデフォルトルート(0.0.0.0/0)とする
2. パブリックサブネットは、インターネットゲートウェイをデフォルトルート(0.0.0.0/0)とする

NATゲートウェイが関わるのは1.だけなので、NATゲートウェイの作成や削除の際にいじる必要があるのはプライベートサブネットのルートテーブルだけということになります。

それでは確認しよう

これから作るLambdaでNATゲートウェイとルーティングを削除してまた作成する際、現時点のシステムの構成と同じものを再現する必要があるため、確認しておきましょう。

① 自身のシステムがどのようなサブネット構成になっているか
確認方法：AWSコンソール > VPC > サブネット

② NATゲートウェイが何個、どのサブネットに置かれているか
確認方法：AWSコンソール > VPC > NATゲートウェイ > 選択 > 詳細 > サブネット

③ 各ルートテーブルのサブネットと0.0.0.0/0（デフォルトルート）のターゲット
確認方法：AWSコンソール > VPC > ルートテーブル > 選択 > サブネットの関連付け、ルート

Lambda実装

それではいよいよLambdaを作成します。
下記のコードは私のシステムに合わせて作成したものです。

途中に待機処理がいくつかありますが、これはNATゲートウェイとルーティングの作成や削除は順番が非常に重要であるためです。前の処理を待ってから次の処理に入る必要があります。

NATゲートウェイ作成Lambda

以下の内容を実装します。

1. 利用可能なElastic IPを取得
2. NATゲートウェイを作成
3. ルーティングを作成

import os
import boto3

# NATゲートウェイが置かれるパブリックサブネットのID
public_subnet = os.environ["PUBLIC_SUBNET_ID"]
# プライベートサブネット１のID
private_subnet1 = os.environ["PRIVATE_SUBNET1_ID"]
# プライベートサブネット２のID
private_subnet2 = os.environ["PRIVATE_SUBNET2_ID"]

client = boto3.client("ec2")


def allocate_Eip():
    """Elastic IPの取得"""
    response = client.allocate_address(Domain="vpc")
    allocation_id = response["AllocationId"]
    # Elastic IPにタグを追加
    client.create_tags(
        Resources=[allocation_id],
        Tags=[
            {
                "Key": "Name",
                "Value": "Elastic IPの名前",
            }
        ],
    )
    return allocation_id


def start_natgw(Eip, subnet):
    """NAT GateWayの開始処理"""
    # NATゲートウェイがすでに存在すればreturn
    filters = [
        {"Name": "subnet-id", "Values": [subnet]},
        {"Name": "state", "Values": ["available"]},
    ]
    response = client.describe_nat_gateways(Filters=filters)
    if response["NatGateways"]:
        return None
    # NATゲートウェイを作成
    response = client.create_nat_gateway(
        AllocationId=Eip,
        SubnetId=subnet,
        TagSpecifications=[
            {
                "ResourceType": "natgateway",
                "Tags": [
                    {
                        "Key": "Name",
                        "Value": "NATゲートウェイの名前",
                    },
                ],
            },
        ],
    )
    natid = response["NatGateway"]["NatGatewayId"]
    client.get_waiter("nat_gateway_available").wait(NatGatewayIds=[natid])
    return natid


def atatch_natgw(natgw, subnet):
    """ルーティング作成"""
    filters = [{"Name": "association.subnet-id", "Values": [subnet]}]
    response = client.describe_route_tables(Filters=filters)
    rtb = response["RouteTables"][0]["Associations"][0]["RouteTableId"]
    client.create_route(
        DestinationCidrBlock="0.0.0.0/0", NatGatewayId=natgw, RouteTableId=rtb
    )


def handler(event, context):
    eip = allocate_Eip()
    # NATゲートウェイの作成はパブリックサブネット
    natgw = start_natgw(eip, public_subnet)
    if natgw is None:
        print("nat gateway has already been started.")
        return
    # ルーティングの作成はプライベートサブネット
    atatch_natgw(natgw, private_subnet1)
    atatch_natgw(natgw, private_subnet2)
    print("start nat gateway in " + public_subnet)

NATゲートウェイ削除Lambda

以下の内容を実装します。

1. ルーティングを削除
2. NATゲートウェイを削除
3. Elastic IPを解放

import os
import boto3
import time

# NATゲートウェイがあるパブリックサブネットのID
public_subnet = os.environ["PUBLIC_SUBNET_ID"]
# プライベートサブネット１のID
private_subnet1 = os.environ["PRIVATE_SUBNET1_ID"]
# プライベートサブネット２のID
private_subnet2 = os.environ["PRIVATE_SUBNET2_ID"]

client = boto3.client("ec2")


def detach_natgw(subnet):
    """ルーティング削除処理"""
    filters = [{"Name": "association.subnet-id", "Values": [subnet]}]
    response = client.describe_route_tables(Filters=filters)
    rtb = response["RouteTables"][0]["Associations"][0]["RouteTableId"]
    client.delete_route(DestinationCidrBlock="0.0.0.0/0", RouteTableId=rtb)
    # ルートテーブルからルートが削除されるまで待機
    while True:
        response = client.describe_route_tables(RouteTableIds=[rtb])
        routes = response["RouteTables"][0]["Routes"]

        if not any(route["DestinationCidrBlock"] == "0.0.0.0/0" for route in routes):
            break

        time.sleep(5)  # 5秒間待機してから再度チェック


def stop_natgw(subnet):
    """NAT GateWayの開始処理"""
    filters = [
        {"Name": "subnet-id", "Values": [subnet]},
        {"Name": "state", "Values": ["available"]},
    ]
    response = client.describe_nat_gateways(Filters=filters)
    # NATを削除する前にElastic IPを取得しておく（Elastic IP解放時に使うため）
    eip = response["NatGateways"][0]["NatGatewayAddresses"][0]["AllocationId"]
    natgw = response["NatGateways"][0]["NatGatewayId"]
    client.delete_nat_gateway(NatGatewayId=natgw)
    # NAT Gateway が削除されるまで待機
    waiter = client.get_waiter("nat_gateway_deleted")
    waiter.wait(NatGatewayIds=[natgw])
    return eip


def handler(event, context):
    # 削除するルーティングはプライベートサブネットのもの
    detach_natgw(private_subnet1)
    detach_natgw(private_subnet2)
    # NATゲートウェイはパブリックサブネットにある
    eip = stop_natgw(public_subnet)
    # Elastic IPを解放
    client.release_address(AllocationId=eip)
    print("stoped nat gateway in " + public_subnet)

終わり

あとはこれをスケジューラとか使って自動実行してやるだけです。

参考記事