top of page
検索
  • 執筆者の写真Naoya Yamashita

Lambdaのロギング方法

はじめに

Pythonには logging という標準モジュールがあり、ロギングにはこれを利用するのが第一候補ですが、独特な実装になっており正しく使うには慣れが必要です。

Lambdaで使いCloudWatch Logsに書き出す前提で、以下のような挙動を実現したいとします。

  • JSON形式で出力する。

  • ボイラーテンプレートを極力なくし、通常のPythonプロジェクトと同じ様に利用できる。

  • 今までのLambda記事(「ローカルテストが行える」「全てのLambdaで同じコードを共有する」)と組み合わせて問題なく使えること。

loggingライブラリの動作把握

loggingの使い方

まず logging ライブラリについて紹介します。公式のドキュメントは以下にあります。

これが logging の利用方法です。 info, warning, error があり、try の except 句では特別に exception が利用できます。

import logging

logger = logging.getLogger(__name__)

logger.info('hello')
logger.warning('hello')
logger.error('hello')

def error_function() -> float:
    return 1 / 0

try:
    error_function()
except Exception:
    logger.exception('hello')

実行すると以下のような出力を得られます。デフォルトではWarning以上のログが出力されます。

$ python poc1.py
[W] hello
[E] hello
[Ex] hello
Traceback (most recent call last):
  File ".../aws-logging-lambda-sample/poc/poc1/poc1.py", line 13, in <module>
    error_function()
  File ".../aws-logging-lambda-sample/poc/poc1/poc1.py", line 10, in error_function
    return 1 / 0
           ~~^~~
ZeroDivisionError: division by zero

loggingの設定(自由フォーマット)

フォーマットを整えるため、以下のように書き換えます。

import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.setLevel(logging.INFO)

formatter = logging.Formatter('[%(levelname)s] %(asctime)s - %(name)s: %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# 以降は同じ

急にコード量が増えました。出力は以下のようになります。

$ python poc2.py
[INFO] 2024-01-15 17:47:20,739 - __main__: [I] hello
[WARNING] 2024-01-15 17:47:20,739 - __main__: [W] hello
[ERROR] 2024-01-15 17:47:20,739 - __main__: [E] hello
[ERROR] 2024-01-15 17:47:20,739 - __main__: [Ex] hello
Traceback (most recent call last):
  File ".../aws-logging-lambda-sample/poc/poc2/poc2.py", line 22, in <module>
    error_function()
  File ".../aws-logging-lambda-sample/poc/poc2/poc2.py", line 19, in error_function
    return 1 / 0
           ~~^~~
ZeroDivisionError: division by zero

loggingの設定(jsonフォーマット)

json形式で出力する場合は logging.Formatter を継承し、自前のフォーマッターを定義する必要があります。

import json
import logging
import traceback
from datetime import datetime
from typing import Any

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.setLevel(logging.INFO)

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        data = (
            {
                'level': record.levelname,
                'message': record.getMessage(),
                'name': record.name,
                'location': f'{record.filename}:{record.lineno}',
                'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            } |
            (
                {
                    'exception': f'{record.exc_info[0].__name__}: {record.exc_info[1]}',
                    'traceback': ''.join(traceback.format_tb(record.exc_info[2])),
                }
                if record.exc_info
                else {}
            )
        )
        extra_data = {
            key: val for key, val in record.__dict__.items() if key.startswith('$')
        }
        return json.dumps(data | extra_data, ensure_ascii=False, default=str)

    def format_exc_info(self, exc_info: tuple[BaseException, BaseException, Any]) -> str:
        return ''.join(traceback.format_tb(exc_info[2]))

formatter = JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)

# 以降は同じ

このフォーマッターを使うことで、実行結果は以下のようになります。

$ python poc3.py
{"level": "INFO", "message": "[I] hello", "name": "__main__", "location": "poc3.py:45", "timestamp": "2024-01-15T21:51:33.132370"}
{"level": "WARNING", "message": "[W] hello", "name": "__main__", "location": "poc3.py:46", "timestamp": "2024-01-15T21:51:33.132437"}
{"level": "ERROR", "message": "[E] hello", "name": "__main__", "location": "poc3.py:47", "timestamp": "2024-01-15T21:51:33.132468"}
{"level": "ERROR", "message": "[Ex] hello", "name": "__main__", "location": "poc3.py:55", "timestamp": "2024-01-15T21:51:33.132493", "traceback": "  File \".../aws-logging-lambda-sample/poc/poc3/poc3.py\", line 53, in <module>\n    error_function()\n  File \".../aws-logging-lambda-sample/poc/poc3/poc3.py\", line 50, in error_function\n    return 1 / 0\n           ~~^~~\n", "exception": "ZeroDivisionError: division by zero"}

json形式でログを出力することができました。

ConfigDictの利用

loggingは前項までのように手続き的に設定する方法とは別に、dict形式の設定を読み込む方法があります。さらに initpy に置くことで、モジュール初期化時に透過的に読み込むことができます。

つまり init.py に以下のように書きます。

import json
import logging
import logging.config
import traceback
from datetime import datetime
from typing import Any


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        data = (
            {
                'level': record.levelname,
                'message': record.getMessage(),
                'name': record.name,
                'location': f'{record.filename}:{record.lineno}',
                'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            } |
            (
                {
                    'exception': f'{record.exc_info[0].__name__}: {record.exc_info[1]}',
                    'traceback': ''.join(traceback.format_tb(record.exc_info[2])),
                }
                if record.exc_info
                else {}
            )
        )
        extra_data = {
            key: val for key, val in record.__dict__.items() if key.startswith('$')
        }
        return json.dumps(data | extra_data, ensure_ascii=False, default=str)

    def format_exc_info(self, exc_info: tuple[BaseException, BaseException, Any]) -> str:
        return ''.join(traceback.format_tb(exc_info[2]))


logger_config = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'json_formatter': {
            '()': JsonFormatter,
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'json_formatter',
        },
    },
    'loggers': {
        '__main__': {
            'handlers': ['console'],
            'level': 'INFO',
        },
    },
}

logging.config.dictConfig(logger_config)

loggingの設定が init.py に移動したため、 poc4.py は以下のようになります。

import logging

logger = logging.getLogger(__name__)

logger.info('[I] hello')
logger.warning('[W] hello')
logger.error('[E] hello')

def error_function() -> float:
    return 1 / 0

try:
    error_function()
except Exception:
    logger.exception('[Ex] hello')

一番最初の例に戻ったことにお気づきでしょうか。つまり利用側では何も気にせず普段通りにログを出すことで、実際にはjson構造化ログを出力できるという状態になっているのです。

モジュールとして実行しないと init.py が実行されないので、起動コマンドが少し変わって python -m poc4.poc4 で起動します。

$ python -m poc4.poc4
{"level": "INFO", "message": "[I] hello", "name": "__main__", "location": "poc4.py:6", "timestamp": "2024-01-15T22:14:11.164466"}
{"level": "WARNING", "message": "[W] hello", "name": "__main__", "location": "poc4.py:7", "timestamp": "2024-01-15T22:14:11.164533"}
{"level": "ERROR", "message": "[E] hello", "name": "__main__", "location": "poc4.py:8", "timestamp": "2024-01-15T22:14:11.164563"}
{"level": "ERROR", "message": "[Ex] hello", "name": "__main__", "location": "poc4.py:16", "timestamp": "2024-01-15T22:14:11.164588", "exception": "ZeroDivisionError: division by zero", "traceback": "  File \".../aws-logging-lambda-sample/poc/poc4/poc4.py\", line 14, in <module>\n    error_function()\n  File \".../aws-logging-lambda-sample/poc/poc4/poc4.py\", line 11, in error_function\n    return 1 / 0\n           ~~^~~\n"}

意図通りに動きました。完璧です。

Lambdaでの利用方法

動作が確認できたのでLambdaで動かしてみましょう。プロジェクトツリーはこの様になっています。

$ tree
.
├── app
   ├── __init__.py
   └── function
       ├── __init__.py
       └── b1
           ├── __init__.py
           └── src
               ├── __init__.py
               ├── main
                  ├── __init__.py
                  ├── add_numbers.py
                  └── multiply_numbers.py
               └── util
                   ├── __init__.py
                   └── lambda_decorator.py
├── poetry.lock
├── pyproject.toml
└── template.yaml

app/function/b1/src/main/add_numbers.py

import logging
from typing import Any

from .. import util

logger = logging.getLogger(__name__)


def error_function() -> float:
    return 1 / 0


@util.lambda_decorator.lambda_decorator
def handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    args = event['args']

    res = args[0] + args[1]


    logger.info('[I] hello')
    logger.warning('[W] hello')
    logger.error('[E] hello')

    try:
        error_function()
    except Exception:
        logger.exception('[Ex] hello')


    return {
        "status": "ok",
        "result": res,
    }

app/function/b1/src/util/lambda_decorator.py

from typing import Any, Callable
import logging

logger = logging.getLogger(__name__)


def lambda_decorator[**P, R](fn: Callable[P, R]) -> Callable[P, R]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        logger.info(f'[lambda_decorator: start] args: {args}')
        res = fn(*args, **kwargs)
        logger.info(f'[lambda_decorator: end] res: {res}')
        return res

    return wrapper

app/function/b1/src/__init__.py

import os
import json
import logging
import logging.config
import traceback
from datetime import datetime
from typing import Any


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        data = (
            {
                'level': record.levelname,
                'message': record.getMessage(),
                'name': record.name,
                'location': f'{record.filename}:{record.lineno}',
                'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            } |
            (
                {
                    'exception': f'{record.exc_info[0].__name__}: {record.exc_info[1]}',
                    'traceback': ''.join(traceback.format_tb(record.exc_info[2])),
                }
                if record.exc_info
                else {}
            )
        )
        extra_data = {
            key: val for key, val in record.__dict__.items() if key.startswith('$')
        }
        return json.dumps(data | extra_data, ensure_ascii=False, default=str)

    def format_exc_info(self, exc_info: tuple[BaseException, BaseException, Any]) -> str:
        return ''.join(traceback.format_tb(exc_info[2]))


info_logger = {
    'handlers': ['console'],
    'level': 'INFO',
}
logger_config = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'json_formatter': {
            '()': JsonFormatter,
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'json_formatter',
        },
    },
    'loggers': {
        '__main__': info_logger,
        'src': info_logger,
    },
}

logging.config.dictConfig(logger_config)

template.yaml

AWSTemplateFormatVersion: "2010-09-09"

Transform: AWS::Serverless-2016-10-31

Parameters:
  Prefix: { Type: String }

Globals:
  Function:
    Runtime: python3.12
    CodeUri: ./app/function/b1
    MemorySize: 128
    Timeout: 60

Resources:
  FunctionB1AddNumbers:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub ${Prefix}-B1AddNumbers
      Handler: src.main.add_numbers.handler

  FunctionB1MultiplyNumbers:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub ${Prefix}-B1MultiplyNumbers
      Handler: src.main.multiply_numbers.handler

初期状態

コンソールから {"args": [1, 2]} の入力で起動できます。起動した結果、Cloudwatch Logにはこの様に記録されています。



ログが2重になっているのを直す

出力結果を見るとログが2重になっています。これはAWSが登録しているデフォルトのloggerがあるためです。これを無効化します。

init.py の末尾に以下の内容を加えます。

if os.environ.get('LAMBDA_TASK_ROOT'):
    # Disable all AWS managed handlers
    logging.getLogger().handlers.clear()

実行してみるとこの様に記録されます。完璧ですね。



まとめ

このブログの方法を使うことで、普段の logging と全く変わらない状態で構造化ログを出力できます。json構造化ログを出力することで、Cloudwatch logクエリで利用しやすくなります。ぜひ使ってみて下さい。

閲覧数:11回0件のコメント

最新記事

すべて表示

データ状態により異なるSQLを実行させたい

はじめに 近頃担当する業務は夜間バッチでのデータ更新処理が多く、特にDWH的にテーブル再構築(TRUNCATE/INSERT)のパターンを多く使用しています。 その中でSQLで処理を組み上げる時、エラー処理などで条件分岐で異なるSQLを実行したくなる事は珍しくありません。 多くのシステムでは呼び出し側でSQLの実行結果を参照し、次に実行するSQLを選択/実行していると思います。 また、SQLだけで

Quick Sightの分析のクロスアカウントコピー

はじめに 今回のブログではQuick Sightで作成した分析をクロスアカウントコピーするための方法を紹介します。 開発環境で作成した分析を本番環境にコピーしたい時などにこの方法が使えるのではないかと思います。 前提 コピー先アカウントとコピー元アカウントで同じ形のデータセットが存在する。 コピー元ではそのデータセットから作られた分析が存在する。 実施環境 今回のブログで紹介する手順はcloud9

Comentarios


bottom of page