AWS Lambda Powertoolsが便利すぎた #serverless #python

aws-lambda-powertools-pythonの実際の使い方をサンプルコードと一緒に紹介します。

アドベントカレンダー2020

#Lambda Powertools

#Python

#サーバーレス

#AWS

岡春奈

2020.12.06

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、クラスメソッドの岡です。
この記事は AWS LambdaとServerless Advent Calendar 2020 の7日目の記事です。

AWS Lambda Powertoolsとは？

Lambdaでの実装をサポートしてくれるライブラリです。現在、ライブラリが提供されているのはPythonとJavaの2つになります。

ちなみに、DAZNからNode.js用の DAZN Lambda Powertools もでています。

動作確認環境

Python: 3.8.5
aws-lambda-powertools-python: 1.8.0
Serverless Framework: 2.15.0

主な機能

Logging
- LambdaのContextを埋め込んだログの構造化
Tracing
- X-Rayでのトレース
Metrics
- CloudWatchのカスタムメトリクスの作成
Utilities
- バリデーションやパラメータ取得などLambdaでよく使う処理をラップしたもの

インストール

$ pip install aws-lambda-powertools

Logging

ログの設定のため、以下2つの環境変数を設定しておきます。

POWERTOOLS_SERVICE_NAME: sample-devio-app
LOG_LEVEL: DEBUG

Lambdaのハンドラに設定する関数に対して、inject_lambda_context のデコレーターを付与するだけで、ログのJSONフォーマット化とContextの埋め込みをやってくれます。

import json
from typing import Any, Dict
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
logger = Logger()


@logger.inject_lambda_context(log_event=True)
def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]:
    logger.debug('debug')
    logger.info('info')
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Success.'})
    }

log_event パラメータをTrueにすることで、eventデータをINFOログで自動的に出力してくれます。

{
    "level": "INFO",
    "location": "decorate:245",
    "message": {
        "limit": 3
    },
    "timestamp": "2020-12-07 08:00:00,000",
    "service": "sample-devio-app",
    "sampling_rate": 0,
    "cold_start": false,
    "function_name": "list_items",
    "function_memory_size": "1024",
    "function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items",
    "function_request_id": "20014cb4-70cb-470d-beb2-839ce13b618e",
    "xray_trace_id": "1-5fcbb424-02fdaa386ca103a26e858647"
}

実際に出力されたeventデータです。構造化されているのでCloudWatch Logs Insightsで簡単にログを探せるようになりますね。

serviceの部分には環境変数 POWERTOOLS_SERVICE_NAME が設定されています。

また、型ヒントを取り入れている場合は utilities のLambdaContextとAPIGatewayProxyEvent(Lambda プロキシ統合イベント)が提供されているので、eventとcontextのアノテーションに使うことができます。

Handlerモジュール以外でログ出力する

Loggerのインスタンス生成時にChildパラメータをTrueに設定します。

from aws_lambda_powertools import Logger
logger = Logger(child=True)

class Auth:

    def verify(self, token: string):
        logger.info(token)
        # 中略
        return result

上記のようにすることでLoggerを初期化せずにコード全体で共通して利用できます。

パラメータを追加する

Loggerをstructure_logs 関数を使うことで、構造化したログに簡単に任意のパラメータを追加できます。試しにAPI Gatewayの認証から渡される principalId をuser_idとしてログにセットします。

def set_logs(self) -> None:
    logger.structure_logs(append=True, user_id=self.principal_id) # principal_id = user_id
    logger.info('test')

def __set_principal_id(self, event: APIGatewayProxyEvent]) -> None:
    self.principal_id = event.get(
        'requestContext',
        {}).get(
        'authorizer',
        {}).get('principalId')

{
    "level": "INFO",
    "location": "set_logs:59",
    "message": "test",
    "timestamp": "2020-12-07 08:00:00,000",
    "service": "sample-devio-app",
    "sampling_rate": 0,
    "cold_start": false,
    "function_name": "list_items",
    "function_memory_size": "1024",
    "function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items",
    "function_request_id": "20014cb4-70cb-470d-beb2-839ce13b618e",
    "user_id": "test_user",
    "xray_trace_id": "1-5fcbb424-02fdaa386ca103a26e858647"
}

"user_id": "test_user"が追加されました。 locationにはログ出力している関数名が出力されます。

pytest

inject_lambda_context デコレータを追加したハンドラーをそのままテストすると構造化の処理でコケてしまうので、pytest実行時には前処理でダミーのcontextを生成して渡してあげる必要があります。

@pytest.fixture
def lambda_context():
    lambda_context = {
        "function_name": "list_items",
        "memory_limit_in_mb": 128,
        "invoked_function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items",
        "aws_request_id": "20b4014c-beb2-839ce70cb-470d-13b618e",
    }

    return namedtuple("LambdaContext", lambda_context.keys())(*lambda_context.values())

def test_list_items(handler, lambda_context):
    test_event = {'test': 'event'}
    handler(test_event, lambda_context) # this will now have a Context object populated

Trace

Tracerを使うことで、aws_xray_sdkを個別に使わずにサブセグメントやアノテーションの定義を簡単に記述できます。（X-Rayの基本的な説明はここでは割愛します）

今回はServerless Frameworkでデプロイしているのでserverless.ymlにX-Rayのトレースの設定を追加していきます。

provider:
  name: aws
  region: ap-northeast-1
  tracing: 
    apiGateway: true
    lambda: true
  iamRoleStatements:
    - Effect: Allow
      Action:
        - logs:CreateLogGroup
        - logs:CreateLogStream
        - logs:PutLogEvents
        - dynamodb:DescribeTable
        - dynamodb:Query
        - dynamodb:Scan
        - dynamodb:GetItem
        - dynamodb:PutItem
        - dynamodb:UpdateItem
        - dynamodb:DeleteItem
        - xray:PutTraceSegments
        - xray:PutTelemetryRecords
      Resource: "*"

handlerに capture_lambda_handler デコレータを追加します。

from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
from aws_lambda_powertools import Tracer
tracer = Tracer()
logger = Logger()

@tracer.capture_lambda_handler
@logger.inject_lambda_context(log_event=True)
def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]:
    logger.debug('debug')
    logger.info('info')
    return {
        'statusCode': 200,
        'body': {'message': 'Success.'}
    }

DynamoDBからデータ取得するLambdaを直実行した場合のTraceです。

アノテーションの追加

トレースをフィルタするアノテーションを追加します。

from aws_lambda_powertools import Tracer
tracer = Tracer()

@tracer.capture_lambda_handler
def handler(event, context):
    tracer.put_annotation(key="PaymentStatus", value="SUCCESS")

TracerはHanderモジュール以外でインスタンス化した場合も既存構成を引き継いでくれます。

メタデータの追加

from aws_lambda_powertools import Tracer
tracer = Tracer()

@tracer.capture_lambda_handler
def handler(event, context):
    ret = some_logic()
    tracer.put_metadata(key="payment_response", value=ret)

並列リクエストのトレース

aiohttp を使った並列リクエストのトレースも可能です。

import asyncio
import aiohttp

from aws_lambda_powertools import Tracer
from aws_lambda_powertools.tracing import aiohttp_trace_config
tracer = Tracer()

async def aiohttp_task():
    async with aiohttp.ClientSession(trace_configs=[aiohttp_trace_config()]) as session:
        async with session.get("https://httpbin.org/json") as resp:
            resp = await resp.json()
            return resp

pytest

ユニットテストの際には環境変数で無効化できます。

＄ POWERTOOLS_TRACE_DISABLED=1 python -m pytest

Utilities

Event Source Data Classes

今回はAPI GatewayのLambdaプロキシ統合を利用する想定なので APIGatewayProxyEvent を指定していますが、他にも以下のイベントソースを指定できます。

import json
from typing import Any, Dict
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
logger = Logger()


@logger.inject_lambda_context(log_event=True)
def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]:
    logger.debug('debug')
    logger.info('info')
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Success.'})
    }

API Gateway Proxy
API Gateway Proxy event v2
CloudWatch Logs
Cognito User Pool
DynamoDB streams
EventBridge
Kinesis Data Stream
S3
SES
SNS
SQS

Parameters

Parametersユーティリティを使って、以下からパラメータを取得してキャッシュまでしてくれます。

SSM Parameter Store
Secrets Manager
DynamoDB

SSM Parameter Storeから取得

from aws_lambda_powertools.utilities import parameters

def handler(event, context):
    value = parameters.get_parameter("/my/parameter")

    values = parameters.get_parameters("/my/path/prefix")
    for k, v in values.items():
        logger.debug(f"{k}: {v}")

（boto3.ssm の get_parameter, get_parameters_by_pathをラップ）

Secrets Managerから取得

from aws_lambda_powertools.utilities import parameters

def handler(event, context):
    value = parameters.get_secret("my-secret")

（boto3.secretsmanager の get_secret_valueをラップ）

DynamoDBから取得

from aws_lambda_powertools.utilities import parameters
dynamodb_provider = parameters.DynamoDBProvider(table_name="my-table")

def handler(event, context):
    value = dynamodb_provider.get("my-parameter")

（boto3.resource.dynamodb の get_itemをラップ）

from aws_lambda_powertools.utilities import parameters

dynamodb_provider = parameters.DynamoDBProvider(
    table_name="my-table",
    key_attr="MyKeyAttr",
    sort_attr="MySortAttr",
    value_attr="MyvalueAttr"
)

def handler(event, context):
    value = dynamodb_provider.get("my-parameter")

（boto3.resource.dynamodb の queryをラップ）