CloudFront+Lambda@EdgeでDynamoDB(MediaConvertの情報)を返す

2022年5月10日 / 最終更新日時 : 2022年6月5日 admin AWS(Amazon Web Services)

メモリ消費が少なく（128MB）短時間（5秒）で終わるプログラムであれば、CloudFront+Lambda@Edge（ビューワーリクエスト）の方がAPI GatewayやALBよりもコストが抑えられそう。
ただ、現時点ではNode.jsとPythonのみで、制限が色々あったり、デプロイが面倒だったりしますが、慣れれば結構良さそうです。
API Gateway+LambdaでDynamoDB(MediaConvertの情報)を返すで作成したDynamoDBを使って試してみました。

エッジ関数に対する制限 – Amazon CloudFront

Lambda関数を作成
IAMロールにポリシー追加
Lambdaコード修正(Hello from Lambda!)
バージョン発行
CloudFrontのビヘイビアを設定
Lambdaコード（API Gateway/ALBとLambda@Edgeの違い）
Lambdaコード（最終版）

Lambda関数を作成

CloudFrontのビヘイビアに設定するのに必要なので、先に作成して、バージョンを発行します。
Lambda@Edgeはバージニア北部(us-east-1)に作成する必要があるので、新しく作成します。
また、環境変数が設定されているとビヘイビア設定時にエラーになるので、API GatewayやELBで使ったLambdaと共存できません。

https://us-east-1.console.aws.amazon.com/lambda/home?region=us-east-1#/functions

関数の作成
	基本的な情報
		関数名: mediaConvertInputFileToJobIdEdgeApi
		ランタイム: Python 3.9
	▼デフォルトの実行ロールの変更
		実行ロール: ●AWS ポリシーテンプレートから新しいロールを作成
		ロール名: mediaConvertEdgeApi
		ポリシーテンプレート: 基本的な Lambda@Edge のアクセス権限 (CloudFront トリガーの場合)
	[関数の作成]

関数の作成
	基本的な情報
		関数名: mediaConvertJobIdToStatusEdgeApi
		ランタイム: Python 3.9
	▼デフォルトの実行ロールの変更
		実行ロール: ●既存のロールを使用する
		ロール名: service-role/mediaConvertEdgeApi
	[関数の作成]

信頼されたエンティティに edgelambda.amazonaws.com を追加するのが手間なので、ポリシーテンプレートを使いました。

ポリシーがゆるゆるなので、気になる場合はポリーシーを変更すると良さそう。
→ CloudWatchにログが出力されなくなったので調べてみた

IAMロールにポリシー追加

LambdaからDynamoDBにアクセスできるようにIAMにポリシーを追加します。

https://us-east-1.console.aws.amazon.com/iamv2/home#/roles

作成したロール(mediaConvertEdgeApi)を選択して、アクセス許可を追加 → ポリシーをアタッチ

ポリシー名
	■AmazonDynamoDBReadOnlyAccess
	[ポリシーをアタッチ]

TODO: DynamoDBのテーブル名で権限を絞る
→ Amazon DynamoDB: 特定のテーブルへのアクセスの許可 – AWS Identity and Access Management

Lambdaコード修正(Hello from Lambda!)

初期設定されているコードだとエラーになるので、statusCodeをstatusに変更します。

import json

def lambda_handler(event, context):
    # TODO implement
    return {
-        'statusCode': 200,
+        'status': 200,
        'body': json.dumps('Hello from Lambda!')
    }

バージョン発行

デプロイ(Deploy)してから、バージョン → 新しいバージョンを発行

$LATEST から新しいバージョンを発行します。
	バージョンの説明: Hello	※空でもOK
	[発行]

関数のARNをCloudFrontのビヘイビアの設定に使います。
コードを変えたら、毎回、デプロイして新しいバージョンを発行する必要があります。
最後の数字がカウントアップされる。

CloudFrontのビヘイビアを設定

今回は、CloudFront（S3オリジン）構築とawsコマンド（S3, CloudFront）メモで作成したディストリビューションを使います。

https://us-east-1.console.aws.amazon.com/cloudfront/v3/home?region=ap-northeast-1#/distributions

対象のディストリビューション(今回はnuxtapp.nightonly.com)を選択して、

ビヘイビア
	[ビヘイビアを作成]

ビヘイビアを作成
	設定
		パスパターン: /api/media/job_id と /api/media/status で2回ビヘイビアを作成
		オリジンとオリジングループ: デフォルトと同じのでOK	※コードで返すので実質未使用
		ビューワー
			ビューワープロトコルポリシー: ●Redirect HTTP to HTTPS
			許可されたHTTPメソッド: ●GET, HEAD, OPTIONS, PUT, POST, PATCH, DELETE
			HTTPメソッドをキャッシュ: ■オプション
		キャッシュキーとオリジンリクエスト
			キャッシュポリシー: CachingDisabled or かなり短いキャッシュポリシーを作成
			レスポンスヘッダーポリシー: 未選択	※CORSヘッダーをコードで設定する為
	関数の関連付け
		ビューワーリクエスト
			関数タイプ: Lambda@Edge
			関数 ARN/名前: バージョン発行した関数のARNをコピペ
			本文を含める: □	※今回は使わないので含めない
	[ビヘイビアを作成]

キャッシュポリシーは、CachingDisabledでも良いのですが、Gzip無効だったり、大量アクセスに対応できないので、かなり短いキャッシュポリシーを作成・設定するのが良さそう。

キャッシュポリシーを作成
	詳細
		名前: CachingLittle
	TTL設定
		最小TTL: 1
		最大TTL: 5	※5秒
		デフォルトTTL: 5
	[作成]

＜上記と同じなので省略＞

動作確認

デプロイ完了まで少し時間が掛かるのでエラーになったら少し時間（数分程度）を置いてから試してみてください。

httpにアクセス → httpsにリダイレクト → レスポンスが表示される

http://nuxtapp.nightonly.com/api/media/job_id
→ https://nuxtapp.nightonly.com/api/media/job_id
"Hello from Lambda!"

http://nuxtapp.nightonly.com/api/media/status
→ https://nuxtapp.nightonly.com/api/media/status
"Hello from Lambda!"

CloudWatch Logs

ログは応答したキャッシュサーバーのリージョンに保存されます。今回は、東京リージョン
コストが増え続けないように保持期間を設定しておきます。

https://ap-northeast-1.console.aws.amazon.com/cloudwatch/home?region=ap-northeast-1#logsV2:log-groups

■ /aws/lambda/us-east-1.mediaConvertInputFileToJobIdEdgeApi
■ /aws/lambda/us-east-1.mediaConvertJobIdToStatusEdgeApi
[アクション] → [保持設定を編集]
保持期間の設定: 3日	※アプリケーションログなので、短めに設定（1日だと短いので3日にしました）
	[保存]

アクセスのあったリージョン毎に設定や確認しなければならないのは難点ですね。
※画面からテストした時のログは、Lambdaを作ったバージニア北部(us-east-1)に保存されます。

Lambdaコード（API Gateway/ALBとLambda@Edgeの違い）

1. statusじゃないと「502 ERROR」になる

Lambdaコード修正(Hello from Lambda!)にも記載しましたが、改めて記載。

statusCodeだとエラーになるので、statusに変更。

def option_response():
-    return { 'statusCode': 200, 'headers': headers }
+    return { 'status': 200, 'headers': headers } # Tips: Lambda@Edgeはstatusじゃないと「502 ERROR」になる

def error_response(code, message):
-    return { 'statusCode': code, 'headers': headers, 'body': json.dumps({ 'success': False, 'message': message }) }
+    return { 'status': code, 'headers': headers, 'body': json.dumps({ 'success': False, 'message': message }) }

def success_response(job_id):
-    return { 'statusCode': 200, 'headers': headers, 'body': json.dumps({ 'success': True, 'job_id': job_id }) }
+    return { 'status': 200, 'headers': headers, 'body': json.dumps({ 'success': True, 'job_id': job_id }) }

-    return { 'statusCode': 200, 'headers': headers, 'body': json.dumps({**result, **info}) }
+    return { 'status': 200, 'headers': headers, 'body': json.dumps({**result, **info}) }

2. 環境変数が使えない

環境変数が設定されているとビヘイビア設定時にエラーになる。

設定せずにコードに記載。

- import os

- ALLOWED_ORIGINS = os.environ['ALLOWED_ORIGINS']
+ ALLOWED_ORIGINS = 'https://.*\.nightonly\.com' # Tips: Lambda@Edgeは環境変数が使えない

3. Lambda@Edgeはus-east-1なので、DynamoDBのリジョンを指定する

DynamoDBは東京リージョンに作成したので、ap-northeast-1を指定。
us-east-1に置いても、近いリージョンのキャッシュサーバーからリクエストされるので、普通にアクセスの多いリージョンに置くのが良さそう。

- dynamodb = boto3.resource('dynamodb')
+ session = boto3.session.Session(region_name = 'ap-northeast-1') # Tips: Lambda@Edgeはus-east-1なので、DynamoDBのリジョンを指定する
+ dynamodb = session.resource('dynamodb')

4. headersはarrayじゃないと「502 ERROR」になる

冗長な感じですが、仕方ないですね。

headers = { # Tips: API GatewayのCORS設定が効かない為
-    'Access-Control-Allow-Origin': '*',
+    'Access-Control-Allow-Origin': [{ 'key': 'Access-Control-Allow-Origin', 'value': '*' }], # Tips: Lambda@Edgeのheadersはarrayじゃないと「502 ERROR」になる
-    'Access-Control-Allow-Methods': 'GET, OPTIONS',
+    'Access-Control-Allow-Methods': [{ 'key': 'Access-Control-Allow-Methods', 'value': 'GET, OPTIONS' }],
-    'Access-Control-Allow-Headers': '*', # Tips: ChromeでCORSエラーになる為
+    'Access-Control-Allow-Headers': [{ 'key': 'Access-Control-Allow-Headers', 'value': '*' }],
-    'Access-Control-Max-Age': '7200', # Tips: LB経由で数値型を渡すと「502 Bad Gateway」になる
+    'Access-Control-Max-Age': [{ 'key': 'Access-Control-Max-Age', 'value': '7200' }],
-    'Content-Type': 'application/json; charset=UTF-8'
+    'Content-Type': [{ 'key': 'Content-Type', 'value': 'application/json; charset=UTF-8' }]
}

5. パラメータの場所が違う

API Gateway: requestContext → http → method
ALB: httpMethod
Lambda@Edge: Records → 0 → cf → request → method

-    if 'elb' in event['requestContext']:
-        method = event['httpMethod']
-    else:
-        method = event['requestContext']['http']['method']
+    method = event['Records'][0]['cf']['request']['method']

API Gateway, ALB: headers → origin
Lambda@Edge: Records → 0 → cf → request → headers → origin → 0 → value

-    if 'origin' in event['headers']:
+    if 'origin' in event['Records'][0]['cf']['request']['headers']:
-        origin = event['headers']['origin']
+        origin = event['Records'][0]['cf']['request']['headers']['origin'][0]['value']

API Gateway, ALB: queryStringParameters → パラメータ名
Lambda@Edge: Records → 0 → cf → request → querystring にそのまま入る（パースしてデコード）

-    if 'queryStringParameters' not in event:
+    query = urllib.parse.parse_qs(event['Records'][0]['cf']['request']['querystring'])
+    if query == {}:
        return error_response(400, 'Not found parameter.')
-    elif 'job_id' not in event['queryStringParameters']:
+    elif 'job_id' not in query:
        return error_response(400, 'Not found parameter job_id.')

6. CloudFrontのエラーページで設定したページが返却される

コードでステータスは400, 422, 200を返していますが、エラーページで設定されているとそのページが返却されます。

削除してあげればOK
※S3にない場合は403が返却されるので、他のオリジンに繋いで該当コードを返さない場合は、設定しても意味ないですね。500でも403になるのは同じ理由なのかも。

https://nuxtapp.nightonly.com/api/media/job_id
{"success": false, "message": "Not found parameter."}

7. 例外エラーは503が返却される

503 ERROR
The request could not be satisfied.

このページが出るのはダサいので、上の502と合わせてエラーページを追加しました。

Lambdaコード（最終版）

最後にそれぞれのコードと正常系のテストイベントを記載しておきます。

mediaConvertInputFileToJobIdEdgeApi

import boto3
import re
import urllib.parse
import json

session = boto3.session.Session(region_name = 'ap-northeast-1') # Tips: Lambda@Edgeはus-east-1なので、DynamoDBのリジョンを指定する
dynamodb = session.resource('dynamodb')
table_to_job_id = dynamodb.Table('MediaConvertInputFileToJobId')

ALLOWED_ORIGINS = 'https://.*\.nightonly\.com' # Tips: Lambda@Edgeは環境変数が使えない

headers = { # Tips: API GatewayのCORS設定が効かない為
    'Access-Control-Allow-Origin': [{ 'key': 'Access-Control-Allow-Origin', 'value': '*' }], # Tips: Lambda@Edgeのheadersはarrayじゃないと「502 ERROR」になる
    'Access-Control-Allow-Methods': [{ 'key': 'Access-Control-Allow-Methods', 'value': 'GET, OPTIONS' }],
    'Access-Control-Allow-Headers': [{ 'key': 'Access-Control-Allow-Headers', 'value': '*' }],
    'Access-Control-Max-Age': [{ 'key': 'Access-Control-Max-Age', 'value': '7200' }],
    'Content-Type': [{ 'key': 'Content-Type', 'value': 'application/json; charset=UTF-8' }]
}

def lambda_handler(event, context):
    print(event)

    # メソッドチェック
    method = event['Records'][0]['cf']['request']['method']
    if method == 'OPTIONS':
        return option_response()
    elif method != 'GET':
        return error_response(404, 'No route matches.')

    # オリジンチェク　# Tips: 存在する場合のみ
    if 'origin' in event['Records'][0]['cf']['request']['headers']:
        origin = event['Records'][0]['cf']['request']['headers']['origin'][0]['value']
        if origin != '' and origin != None:
            allowd_origins = ALLOWED_ORIGINS.split()
            print({ 'origin': origin, 'allowd_origins': allowd_origins })

            allowd = False
            for allowd_origin in allowd_origins:
                if re.fullmatch(allowd_origin, origin):
                    allowd = True
                    break
            if not allowd:
                return error_response(422, 'ORIGIN is invalid.')

    # パラメータチェック
    query = urllib.parse.parse_qs(event['Records'][0]['cf']['request']['querystring'])
    if query == {}:
        return error_response(400, 'Not found parameter.')
    elif 'input_file' not in query:
        return error_response(400, 'Not found parameter input_file.')

    input_file = query['input_file'][0]
    print({ 'input_file': input_file })

    if input_file == '' or input_file == None:
        return error_response(422, 'Not exist parameter input_file.')

    # DynamoDBから出力情報取得
    response = table_to_job_id.get_item(
         Key = {
            'InputFile': input_file
        }
    )
    print(response)

    if 'Item' not in response:
        return error_response(422, 'Not found item.')

    return success_response(response['Item']['JobId'])

def option_response():
    return { 'status': 200, 'headers': headers } # Tips: Lambda@Edgeはstatusじゃないと「502 ERROR」になる

def error_response(code, message):
    return { 'status': code, 'headers': headers, 'body': json.dumps({ 'success': False, 'message': message }) }

def success_response(job_id):
    return { 'status': 200, 'headers': headers, 'body': json.dumps({ 'success': True, 'job_id': job_id }) }

テストイベント

{
    "Records": [
        {
            "cf": {
                "request": {
                    "headers": {
                        "origin": [
                            {
                                "value": "https://app.nightonly.com"
                            }
                        ]
                    },
                    "method": "GET",
                    "querystring": "input_file=Big+Buck+Bunny.mp4"
                }
            }
        }
    ]
}

mediaConvertJobIdToStatusEdgeApi

import boto3
import re
import urllib.parse
import json

session = boto3.session.Session(region_name = 'ap-northeast-1') # Tips: Lambda@Edgeはus-east-1なので、DynamoDBのリジョンを指定する
dynamodb = session.resource('dynamodb')
table_to_status = dynamodb.Table('MediaConvertJobIdToStatus')

ALLOWED_ORIGINS = 'https://.*\.nightonly\.com' # Tips: Lambda@Edgeは環境変数が使えない

headers = { # Tips: API GatewayのCORS設定が効かない為
    'Access-Control-Allow-Origin': [{ 'key': 'Access-Control-Allow-Origin', 'value': '*' }], # Tips: Lambda@Edgeのheadersはarrayじゃないと「502 ERROR」になる
    'Access-Control-Allow-Methods': [{ 'key': 'Access-Control-Allow-Methods', 'value': 'GET, OPTIONS' }],
    'Access-Control-Allow-Headers': [{ 'key': 'Access-Control-Allow-Headers', 'value': '*' }],
    'Access-Control-Max-Age': [{ 'key': 'Access-Control-Max-Age', 'value': '7200' }],
    'Content-Type': [{ 'key': 'Content-Type', 'value': 'application/json; charset=UTF-8' }]
}

def lambda_handler(event, context):
    print(event)

    # メソッドチェック
    method = event['Records'][0]['cf']['request']['method']
    if method == 'OPTIONS':
        return option_response()
    elif method != 'GET':
        return error_response(404, 'No route matches.')

    # オリジンチェク　# Tips: 存在する場合のみ
    if 'origin' in event['Records'][0]['cf']['request']['headers']:
        origin = event['Records'][0]['cf']['request']['headers']['origin'][0]['value']
        if origin != '' and origin != None:
            allowd_origins = ALLOWED_ORIGINS.split()
            print({ 'origin': origin, 'allowd_origins': allowd_origins })

            allowd = False
            for allowd_origin in allowd_origins:
                if re.fullmatch(allowd_origin, origin):
                    allowd = True
                    break
            if not allowd:
                return error_response(422, 'ORIGIN is invalid.')

    # パラメータチェック
    query = urllib.parse.parse_qs(event['Records'][0]['cf']['request']['querystring'])
    if query == {}:
        return error_response(400, 'Not found parameter.')
    elif 'job_id' not in query:
        return error_response(400, 'Not found parameter job_id.')

    job_id = query['job_id'][0]
    print({ 'job_id': job_id })

    if job_id == '' or job_id == None:
        return error_response(422, 'Not exist parameter job_id.')

    # DynamoDBからステータス取得
    response = table_to_status.get_item(
         Key = {
            'JobId': job_id
        }
    )
    print(response)

    if 'Item' not in response:
        return error_response(422, 'Not found item.')

    return success_response(response['Item'])

def option_response():
    return { 'status': 200, 'headers': headers } # Tips: Lambda@Edgeはstatusじゃないと「502 ERROR」になる

def error_response(code, message):
    return { 'status': code, 'headers': headers, 'body': json.dumps({ 'success': False, 'message': message }) }

def success_response(item):
    job_status = item['JobStatus']
    result = { 'success': True, 'job_status': job_status, 'progress_rate': int(item['ProgressRate']) }
    info = {}
    if job_status == 'COMPLETE':
        info = { 'duration_ms': int(item['DurationMs']), 'width_px': int(item['WidthPx']), 'height_px': int(item['HeightPx']),
                 'output_path': item['OutputPath'], 'playlists': item['Playlists'] }
    elif job_status == 'ERROR':
        info = { 'error_code': int(item['ErrorCode']), 'error_message': item['ErrorMessage'] }

    return { 'status': 200, 'headers': headers, 'body': json.dumps({**result, **info}) }

テストイベント

{
    "Records": [
        {
            "cf": {
                "request": {
                    "headers": {
                        "origin": [
                            {
                                "value": "https://app.nightonly.com"
                            }
                        ]
                    },
                    "method": "GET",
                    "querystring": "job_id=c123456789012-smb6o7"
                }
            }
        }
    ]
}

カテゴリー: AWS(Amazon Web Services)

タグ: API CloudFront CloudWatch CORS DynamoDB Lambda Python