はじめに
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 zerologgingの設定（自由フォーマット）
フォーマットを整えるため、以下のように書き換えます。
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 zerologgingの設定（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.yamlapp/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 wrapperapp/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があるためです。これを無効化します。
if os.environ.get('LAMBDA_TASK_ROOT'):
# Disable all AWS managed handlers
logging.getLogger().handlers.clear()実行してみるとこの様に記録されます。完璧ですね。
まとめ
このブログの方法を使うことで、普段の logging と全く変わらない状態で構造化ログを出力できます。json構造化ログを出力することで、Cloudwatch logクエリで利用しやすくなります。ぜひ使ってみて下さい。
留言