AWS FargateとAPI Gatewayでイベントドリブンな構成を実装する

はじめに

前回までの検証でなんとかPythonのプログラムをFargateで実行するところまで構築した。
↓

前回の内容ではAWSコンソールから手動でTaskをRunしていたが、今回はどうしてもPostmanから実行したかったのでその内容を実装した記録。

ちなみにここまでで以下のような構成を検証がてら作成してきたが、今回は赤枠の部分のみの記録。本投稿の内容を把握するのに、前回までの構成を知っておく必要は特にない。

悩み

今回はPostmanからAPIリクエストを実施して、Fargateタスクを実行するようなイベントドリブンな構成を構築したかったが、その際に以下のようなポイントで頭を悩ませたのでメモ。

• 個人の検証用なのでなるべく料金を抑えたい。※結局ここが重要！（笑）
• 色んな構成の検証はしたいが、検証後にも構築後の構成を維持して少しずつ拡張していきたい。→ 維持費が安いに越したことはない！
• Fargateは使いたい。API Gatewayも使いたい。
• Fargateをサービスタイプにてデプロイすると、月額料金が割とかかってしまう。→イベントドリブンにしたい。（常駐サービスである必要はない。）
• API GatewayからFargateに対して直接Rest APIを投げるカタチにしたいが、素直にその構成を採用するとVPCエンドポイント（privatelink）のサービス利用が必要で、微小ではあるが月額コストがかかってくる。
• APIリクエストボディにセットされる文字列（DynamoDBにセットする値）はFargateに渡したい。
• Fargateをサービスと起動させてAPIバックエンドのような方式も検証したい（がそのためにはVPCエンドポイントやFargateの維持費などコストがかかってきてしまう。。→この検証はまた次の機会に！）

↓以下結論。
API Gatewayで構築済のREST APIは維持したまま、イベントドリブンでFargateタスクを実行して最小コストで検証環境を維持していきたい。
⇒ API Gateway ~ Lambda ~ Fargateの構成にする。

ちなみにFargateの活用方法や、必要なその他AWSサービス、コストなどを比較する際に以下のサイトを参考にした。

今回は以下の手順で実際に構築。

1. Fargateで実行するpythonコードを修正。
2. API Gatewayから呼び出されるLambdaを修正（もしくは新規作成）。
3. API Gateway側の設定の確認（特に変更は不要だった）。
4. 動作確認

実施内容

Fargateでの実行コードを修正 & イメージプッシュ

以下のようにpythonコード記載。
os.environ["EVENT_MESSAGE"]から環境変数に渡されるメッセージ内容（Lambda側でセット）を取得している。

topic_publisher.py

import base64
import boto3
from datetime import datetime
import json
import os
from os import path

EVENT_MESSAGE = os.environ["EVENT_MESSAGE"]

client = boto3.client('sns', region_name='ap-northeast-1')

def logging(errorLv, funcName, errorMsg):
    loggingDateStr=(datetime.now()).strftime('%Y%m%d %H:%M:%S')
    print(loggingDateStr + " " + funcName + " " + "[" + errorLv + "] " + errorMsg)
    return

def main():
    
    logging("info", "publisher fargate", "lambda started")
    print('fargate started')
    
    params = {
        'TopicArn': 'arn:aws:sns:ap-northeast-1:123456789012:myFirstTopic_std',
        'Subject' : 'Published From: deliverTopicToSNS_Test01',
        'Message' : EVENT_MESSAGE
    }
    
    try:
        response = client.publish(
            TopicArn = params['TopicArn'],
            Subject = params['Subject'],
            Message = params['Message']
        )
        
        print(json.dumps(response))
        return {
            'statusCode': 200,
            'body': json.dumps('A new message has been published')
        }
        
    except Exception as e:
        print(e)
        raise e
    
main()

続いてコンテナイメージをビルドしてECRレポジトリにpushしていく。
ECRにログインして、

イメージをビルドし、

タグ付けしてプッシュ

使用しているDokerfileやrequirement.txtなどは前回分から差分はないので、記事冒頭のリンクから確認可能。Task定義も前回の設定から変更なし。

API Gatewayから呼び出されるLambdaを修正

続いてAPI Gatewayから呼び出され、ECSのTaskをRunするLambdaを作成していく。元々存在していたSNS Topicを配信するLambdaを修正して利用したので「修正」となっているが、新規作成でも全然問題ない。
ECSタスク起動時に渡したい引数などをLambdaの環境変数にセットしている。

lambda_function.py

import boto3
from datetime import datetime
import json
import os

ecs_client = boto3.client("ecs")

ECS_CLUSTER = os.environ["ECS_CLUSTER"]
TASK_DEFINITION = os.environ["TASK_DEFINITION"]
SUBNET_ID_1 = os.environ["SUBNET_ID_1"]
SUBNET_ID_2 = os.environ["SUBNET_ID_2"]

def logging(errorLv, LambdaName, errorMsg):
    loggingDateStr=(datetime.now()).strftime('%Y%m%d %H:%M:%S')
    print(loggingDateStr + " " + LambdaName + " " + "[" + errorLv + "] " + errorMsg)
    return

def lambda_handler(event, context):
    logging("info", context.function_name, "lambda started")
    message = event['body']
    
    try:
        response = ecs_client.run_task(
            cluster = ECS_CLUSTER,
            launchType = "FARGATE",
            platformVersion = "1.3.0",
            networkConfiguration = {
                "awsvpcConfiguration": {
                    "subnets": [SUBNET_ID_1, SUBNET_ID_2],
                    "assignPublicIp": "ENABLED",
                }
            },
            overrides = {
                "containerOverrides": [
                    {
                        "name": "countainer-pub",
                        "environment": [
                            {"name": "EVENT_MESSAGE", "value": message}    
                        ]   ,
                    },
                ],
            },
            taskDefinition = TASK_DEFINITION,
        )
        
        print(response)
        return {
            'statusCode': 200,
            'body': json.dumps('A new message has been published')
        }
        
    except Exception as e:
        print(e)
        raise e
           

Lambdaの環境変数は以下のように設定している。

run_taskの引数に設定できる内容などは以下の公式Docを参照。

Lambdaに設定している実行ロールに必要なポリシーを追加する。
まずはecs:RunTask。

次にiam:PassRole。

このようになる。

API Gateway側の設定の確認

特に変更は必要なかったがどんな設定がしてあったのか確認しておく。

POSTリクエストを構成しており、LAMBDA_PROXY統合で上記Lambdaを呼び出している。

動作確認

Postmanからリクエストを実施。

無事にステータス200の返却を確認できた。

ECS側の画面も見てみるとTaskがRunningになっていることが確認できた。無事にLambda経由でFargateタスクが動いたっぽい！

後続の処理でDynamoDBに新規ItemをPutしている処理があるので、DynamoDBテーブルも確認してみる。ここも無事に処理が完了されていた。

おわりに

今回はコストをかけたくなかったのと、これまでの検証用の構成を引き継いだカタチだったのでこのような構成になった。実際にはAPI GatewayでコールされるFargateタスクはこのようなTaskタイプではなくサービスとして展開されているケースの方が多いのだろうと思う。（色々ググってみてもそちらの構成記事の方が多かった。）Fargateをイベントドリブンとしてだけでなく、API Backendとして活用しているような構成の検証も今後実施していきたい。

参考サイト