TL:DR

AWSでサーバレス開発を初めて早1年強、毎回関数コードにLoggerやX-Rayのコードを埋め込むのに疲れ始めていました。
そんな折に発見したLambda Powertools（Python）が非常にお手軽かつ便利そうだったので使ってみました。

AWS Lambda Powertoolsって何？

GitHubのProjectより引用

A suite of Python utilities for AWS Lambda functions to ease adopting best practices such as tracing, structured logging, custom metrics, and more. (AWS Lambda Powertools Java and Typescript is also available).

上記からも分かるように、Lambda関数実行時のトレーシング、ロギングをベストプラクティスに沿う形で簡単に導入できるユーティリティスイートとなっています。

導入してみる

Powertoolsは公開Lambda LayerやPyPiで提供されているので、SAMやCFnテンプレートに埋め込む、requirements.txtに記述する等、様々な方法でデプロイすることが可能となっています。

筆者はSAMでデプロイすることが多いので、今回はSAMテンプレートに仕込んでみます。
といっても、Layerを指定するだけなので、ハイライトしている部分の2行で公開Layerを指定するだけでOKです。楽ちん。

今回はX-Rayも有効化してトレーシングするので、SAMテンプレート内にIAMロールの定義も仕込んでいます。

template.yml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Globals:
  Function:
    Timeout: 60
    Tracing: Active
  Api:
    TracingEnabled: True

Resources:
  LambdaRole:
    Type: AWS::IAM::Role
    Properties:
      Path: /
      RoleName: 'sam-test-app-role'
      AssumeRolePolicyDocument:
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - "lambda.amazonaws.com"
            Action:
              - "sts:AssumeRole"
      MaxSessionDuration: 3600
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
        - 'arn:aws:iam::aws:policy/AWSXrayWriteOnlyAccess'

  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
      Role: !GetAtt LambdaRole.Arn
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.9
+     Layers:
+       - !Sub arn:aws:lambda:${AWS::Region}:017000801446:layer:AWSLambdaPowertoolsPython:19
      Architectures:
        - x86_64
      Environment:
        Variables:
          LOG_LEVEL: INFO
          POWERTOOLS_SERVICE_NAME: sam-test-app-lambda
      Events:
        HelloWorld:
          Type: Api
          Properties:
            Path: /hello
            Method: get

Powertoolsで提供されるユーティリティ

Powertoolsでは様々なユーティリティが提供されていますが、本記事では代表的な3つのユーティリティ（Core Utilities）を紹介します。

Tracer

X-Rayの軽量ラッパーとなっており、Lambdaのコールドスタートや例外をメタデータとしてキャプチャすることが可能となっています。

LayerからインポートしたTracerクラスをインスタンス化して、@tracer.capture_lambda_handlerデコレータをLambdaハンドラの頭に追記してあげます。

import json

+ from aws_lambda_powertools import Tracer
+ tracer = Tracer()

+ @tracer.capture_lambda_handler
def lambda_handler(event, context):
    return {
        "statusCode": 200,
        "body": json.dumps({
            "message": "hello world",
        }),
    }

上記の状態でAPI GatewayからLambdaをキックすると下記のようなX-Rayのトレースが生成されます。

特に、本ユーティリティクラスで嬉しいところはX-Rayのトレースにアノテーションを簡単に追加できる点ではないでしょうか？例えば、try~except文において例外が発生した場合のトレースをドリルダウンして分析したい場合、下記のようなコードに修正してあげればX-Rayコンソールから簡単に追跡することが可能です。

app.py

import json
+ from aws_lambda_powertools import Tracer
+ tracer = Tracer()

+ @tracer.capture_lambda_handler
def lambda_handler(event, context):
    try:
+       tracer.put_annotation(key="isSuccess",value="True")
        return {
            "statusCode": 200,
            "body": json.dumps({
                "message": "hello world",
            }),
        }
    except Exception as e:
+       tracer.put_annotation(key="isSuccess",value="False")
        return {
            "statusCode": 500,
            "body": json.dumps({
                "message": e,
            }),
        }

Logger

JSONで構造化された文字列をログとして吐き出してくれます。
先程のTracerと同じノリでインスタンス化しています。

app.py

import json
+ from aws_lambda_powertools import Logger
+ logger = Logger()

+ @logger.inject_lambda_context
def lambda_handler(event, context):
+   logger.append_keys(appended_key="appended_key")
+   logger.info("hello from powertool!!")
    return {
        "statusCode": 200,
        "body": json.dumps({
            "message": "hello world",
        }),
    }

上記のコードのlogger.info("hello from powertool!!")部分でログを出力しています。
実際にCloudWatch Logsに吐き出されたログを捕まえると下記のようなJSONになっています。

messageカラムに指定した文字列が出力されているのはもちろんですが、cold_startやtimestampといったメタデータが自動的に付与されてJSONとして構造化されて視認性が非常に良いことが分かります。
各カラムにどのような値が出力されているのかは公式ドキュメントに記載があるので参考にしてみてください

Metrics

CloudWatch Metricsでカスタムメトリクスを作成することができます。

app.py

import json
+ from aws_lambda_powertools import Metrics
+ from aws_lambda_powertools.metrics import MetricUnit

+ metrics = Metrics(namespace="test-sam-app", service="lambda-powertool")

+ @metrics.log_metrics
def lambda_handler(event, context):
    try:
+       metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        return {
            "statusCode": 200,
            "body": json.dumps({
                "message": "hello world",
            }),
        }
    except Exception as e:
+       metrics.add_metric(name="FailCount", unit=MetricUnit.Count, value=1)
        return {
            "statusCode": 500,
            "body": json.dumps({
                "message": e,
            }),
        }

上記のコードを動かすと、CloudWatch Metricsにカスタムネームスペースが作成され、メトリクスが記録されていることが分かります。（添付のコードだと、関数が正常終了/例外終了した際にそれぞれカウントを＋1する実装となっています）

まとめ

偶然発見したLambda Powertoolsを試してみましたが、ロギング、トレーシングが非常に簡単に実装できて素晴らしいなと思いました。
今までだとboto3書いて〜、import loggingして〜みたいなことをする必要があったので、コードも非常にスッキリしたような気がします。
（ログのフォーマット標準化も必要ありませんし、、）

今回は紹介できませんでしたが、SSMパラメータストアやSecret Managerからパラメータを取得するユーティリティや、SNSやSQSからのイベントハンドラもユーティリティとして提供されているようなので、Lambdaで処理を書く際には今後導入していこうと思います。

2022/05時点で正式サポートされているランタイムはPythonとJavaの2つとなります。
（TypeScript版も利用可能ですが、まだβ版とのことなので利用の際はご注意ください）

AWS Lambda Powertoolsで実現する快適Lambda生活

TL:DR

AWS Lambda Powertoolsって何？

導入してみる

Powertoolsで提供されるユーティリティ

Tracer

Logger

Metrics

まとめ