Pythonのstructlogを使いこなす
記事の内容
- Python言語における
structlogの紹介 -
structlogの使い方の解説 - 私の設定テンプレートの共有と解説
モチベーション
-
structlogの使い方に関する情報がネットにないので書く -
structlogの作者への寄付金が増えてほしい、という祈り -
structlogがPythonのloggerツールの中で一番使い勝手がよい、という宗教戦争
structlog について
structlog はPythonのログを改善するライブラリーであり、一貫性のある簡易なAPIと 構造化ロギング（Structured Logging） を採用している点に特徴がある。
作者はattrsの作者として有名なhynek(Hynek Schlawack)。[1]
構造化ロギングについて
structlogが実現している 構造化ロギング(Structured logging) とは何かについて、公式ドキュメントでは以下のように説明されている。
Structured logging means that you don’t write hard-to-parse and
hard-to-keep-consistent prose in your log entries.
Instead, you log events that happen in a context of key-value pairs.
# 構造化ロギングではログ(記録)を、従来の解析し難く一貫性も保てない
# 散文形式の記法に代わり、起きたイベントをキーと値のペアの組み合わせの形式でログにします。
structlog を使うモチベーション
公式ドキュメントの主張
structlogの利点は公式ドキュメントの冒頭に記載されている。
以下は公式ドキュメントの冒頭とそれを翻訳したもの。
structlog is the production-ready logging solution for Python:
# structlogは本番環境でも安定して動くPythonのロギングの解決策です
# 『シンプルさ』、『パワフルさ』、『実行速度』に利点があります
Simple: At its core, everything is about functions that take and
return dictionaries – all hidden behind familiar APIs.
# シンプルさ
# 機能の中心は辞書データの受け渡しの関数であり、Pythonの標準的なAPIで操作できます
- Powerful: Functions and dictionaries aren’t just simple,
they’re also powerful. structlog leaves you in control.
# パワフルさ
# 関数と辞書のインターフェスはシンプルでありながら、高機能で拡張性を備えています
# structlogでは、細かな挙動に至るまですべてをユーザーが制御できます
Fast: structlog is not hamstrung by designs of yore.
Its flexibility comes not at the price of performance.
# 実行速度
# structlogは、Pythonの古いロギングシステムの制約に縛られておらず、融通が利きます
# しかも、それは速度性能を犠牲にすることなく実現されています
APIの設計に対しての私見
構造化ロギングは、その明確性と一貫性により、ログ解析やトラブルシューティングを飛躍的に効率化する。Python標準のloggingモジュールでも冗長な手続きを踏めば構造化ロギングは実現できるが、structlogは、構造化ロギングに特化したAPIを提供しており、開発者の負担が軽減される。
structlogの革新性は、従来のロギングが抱えていた「高機能なログの実現」と「使いやすさ」のジレンマを克服した点にある。
従来の構造化ロギングは各要素への詳細なパラメーター設定を必要とし、それはコードの複雑化を招いた。ログの内容を作り込むほどコードの複雑性はべき乗に膨れ上がっていき、扱い易さは損なわれた。しかし、structlogは直感的で操作性のまま、ユーザー側の高度な要求を簡単に実現できる。
このノベーションは作者hynekの卓越したAPI設計と高度な実装能力によって実現された。その他に実行速度のメリットもある。全体を評するなら作者の主張通り「シンプル、パワフル、ファスト」の一石三鳥の評価が妥当だろう。
他ツールとの比較
1. Pythonのloggingモジュール vs structlog
- loggingモジュールの利点
- Pythonのビルトインであるため、ライブラリを導入する必要がない
- loggingモジュールの欠点
- 設計がJavaのライブラリの移植でAPIのデザインがPythonと一貫性がない
- そのためドキュメントを解読するのに数日かかる
- 正しく使えるようになるには一週間以上、長ければ年単位の時間を要する
2. logru vs structlog
- logruの利点
- 導入方法がシンプルで最初のlogを成功させるまでは最速
- 対してstructlogは使い始めにドキュメントを一時間くらい読む必要がある
- logruの欠点
- loggingモジュールの薄いラッパーであり深入りするなら、loggingモジュールと変わらない
- structlogはloggingモジュールのラッパーとして扱え、
その場合でもシンプルなstructlogのAPIが使えるが、logruはそうとは限らない
本題
導入の敷居が高いのがstructlogの欠点である。
logruはドキュメントのコードをコピペするだけで動くが、structlogには簡単なチュートリアルがなくドキュメントも難しい。[4]
そのため、本記事で簡単なstructlogのチュートリアルと私のstructlogのテンプレートを共有する。
チュートリアル
この入力と出力を読めば基本的な使い方は習得できる
チュートリアルのコード部分
import structlog
# 最初はlogのオブジェクトを呼び出すことから始める
# bindを使い、logをキーと値の組合せで構造化している。キーも値も任意の入力が通る
mylog = structlog.get_logger().bind(purpose="Tutorial")
# lesson="one"でinfoの内容を構造化している
# bindで構造化された内容はunbindされない限り、保持され続ける
# "hello, world"およびlesson="one"のコンテキストは次のログには保持されない。(一度きり)
mylog.info("hello, world", lesson="one")
# 複数の構造を持たせることができる
mylog.info("good bye", lesson="two", feel="sad")
# 複数のlogを定義し並列で使うことができる
second_log = structlog.get_logger().bind(target="multi_log_tutorial")
# 深刻度に応じて出力形式、debug, info, warning, error, criticalを使い分ける
# https://docs.python.org/ja/3/library/logging.html#logging-levels
second_log.warn("it is second logger")
# 文字列型以外も構造に入れられる
second_log.error("Error Trial", is_danger=False)
# second_logとmylogは共存できる
mylog.info("First log is still alive", lesson="last")
私のstructlogテンプレート
上のチュートリアルではstructlogの初期設定の挙動をそのまま使用している。
structlogには挙動の変更の余地が無限に残されている。また公式推奨のオススメ設定のようなものはない。
そこで設定の一例として私が使用している設定の組み合せ(以下、テンプレート)を共有し、その動かし方と仕様を解説する。
"""logger_config.py: This module configures the logging.
This module uses structlog to configure logging.
"""
import logging
import sys
import structlog
from structlog.dev import ConsoleRenderer
from structlog.processors import JSONRenderer
def configure_logger():
"""
Configures the global logging settings for the application using structlog and Python's built-in logging module.
This function applies various processors to the logs, including adding log level, logger name, and timestamp,
rendering stack information, decoding unicode, and wrapping the message for formatter. The logs are output in
both console and a file named "application.log". Console logs are formatted for readability while the file logs are
in JSON format for further processing.
It is recommended to call this function only once at the start of the application.
Notes
-----
The log level is set to INFO, meaning that INFO and above level logs (WARNING, ERROR, CRITICAL) are recorded.
The configuration includes a `cache_logger_on_first_use` setting to improve performance by caching the logger
on its first use.
The standard output (stdout) and a file "application.log" are used as log outputs. If "application.log"
already exists, new logs will be appended to the existing file.
"""
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M.%S", utc=False),
structlog.processors.StackInfoRenderer(),
structlog.processors.UnicodeDecoder(),
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
handler_stdout = logging.StreamHandler(sys.stdout)
handler_stdout.setFormatter(structlog.stdlib.ProcessorFormatter(processor=ConsoleRenderer()))
handler_file = logging.FileHandler("application.log")
handler_file.setFormatter(structlog.stdlib.ProcessorFormatter(processor=JSONRenderer()))
root_logger = logging.getLogger()
root_logger.addHandler(handler_stdout)
root_logger.addHandler(handler_file)
root_logger.setLevel(logging.INFO)
動かし方
まず、上の内容のファイルをlogger_configとして保存する。そしてmain.pyのスクリプトを以下のように作成し実行する。
from logger_config import configure_logger
configure_logger()
log = structlog.get_logger()
log.info("hello, world")
get_logger()をする前にconfiguer_logger()を呼ぶことによってget_logger()の処理がカスタマイズされたloggerを呼ぶようになる。
テンプレート解説
私がこのテンプレートで実現したこと
- 目的1: ターミナル上でlogを綺麗に出力すること
- 目的2: ロギングの出力の結果のログを
aplication.logにjson形式で保存すること
達成しないこと
- クラウドサービスのログ管理ツールとの連携
- ELKスタックなどの中央ログ管理ツールとの連携
- ロギング処理の高速化・並列処理の実装
- ログの圧縮保存
この記事ではこれらの要素について言及しないが公式ドキュメントには関連する記載がある。[5]
詳細解説
handler_stdout = logging.StreamHandler(sys.stdout)
handler_stdout.setFormatter(structlog.stdlib.ProcessorFormatter(processor=ConsoleRenderer()))
handler_file = logging.FileHandler("application.log")
handler_file.setFormatter(structlog.stdlib.ProcessorFormatter(processor=JSONRenderer()))
root_logger = logging.getLogger()
root_logger.addHandler(handler_stdout)
root_logger.addHandler(handler_file)
Pythonの公式ドキュメントに記載のあるPythonのloggerの専門用語やオプションの解説はせず、structlog特有のオプションにフォーカスする。
目的1 と 目的2 の実現のためにそれぞれ handler_stdout と handler_file をstructlog上で用意し、それをPythonのloggerモジュールに渡す。
そのhandler_stdout (ターミンル出力のハンドラ)には structlog.dev.ConsoleRenderer を handler_file (ファイル出力のハンドラ)には structlog.processors.JSONRenderer をセットする。
ConsoleRenderer はターミナル上で視認性の高いログを表示する。
JSONRenderer はログをapplication.logにjson形式で保存する。
拡張
以下のように引数を作って使うことも多い。
logging.LEVELの実態は整数であるが、私はその対応関係を覚えていない、かつ、数値のためだけにloggingをimportしたくないので文字列で呼べるようにしている。なおJavaと違い、CRITICALとFATALは同列である。
def configure_logger(
logging_level: int
| Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "FATAL"] = logging.INFO,
) -> None:
root_logger = logging.getLogger()
match logging_level:
case int():
root_logger.setLevel(logging_level)
case "DEBUG":
root_logger.setLevel(logging.DEBUG)
case "INFO":
root_logger.setLevel(logging.INFO)
case "WARNING":
root_logger.setLevel(logging.WARNING)
case "ERROR":
root_logger.setLevel(logging.ERROR)
case "CRITICAL", "FATAL":
root_logger.setLevel(logging.CRITICAL)
case _:
root_logger.setLevel(logging.INFO)
テンプレートの使用例
以下はテンプレートを使用した構造化ロギングの例である。
解説は後述する。
import asyncio
import uuid
import structlog
from logger_config import configure_logger
from strategies import StrategyA
from strategies import StrategyB
from strategies import StrategyC
async def main():
configure_logger()
run_id = str(uuid.uuid4())
log = structlog.get_logger(__name__).bind(run_id=run_id)
strategy_a = StrategyA(run_id)
strategy_b = StrategyB(run_id)
strategy_c = StrategyC(run_id)
try:
await asyncio.gather(
strategy_a.execute(),
strategy_b.execute(),
strategy_c.execute(),
)
except Exception as e:
log.exception("An error occurred inside asyncio gather", exception=str(e))
log.info("Application finished.")
if __name__ == "__main__":
asyncio.run(main())
import asyncio
import inspect
import structlog
class StrategyA:
''' 一番シンプルな実装 '''
def __init__(self, run_id):
self.run_id = run_id
self.log = structlog.get_logger(__name__).bind(
run_id=self.run_id,
strategy_name=self.__class__.__name__,
)
async def execute(self):
self.log.info("Executing strategy A...")
await asyncio.sleep(2)
self.log.info("Strategy A execution finished.")
class StrategyB:
''' 実行時の詳細ログを表示する '''
def __init__(self, run_id):
self.run_id = run_id
self.log = structlog.get_logger(__name__).bind(
run_id=self.run_id,
strategy_name=self.__class__.__name__,
)
async def execute(self):
self.log.info(
"Executing strategy B...", function_name=inspect.currentframe().f_code.co_name
)
await asyncio.sleep(3)
self.log.info(
"Strategy B execution finished.", function_name=inspect.currentframe().f_code.co_name
)
class StrategyC:
''' 必ず失敗する '''
ERROR_MESSAGE = "Something went wrong in StrategyC"
def __init__(self, run_id):
self.run_id = run_id
self.log = structlog.get_logger(__name__).bind(
run_id=self.run_id,
strategy_name=self.__class__.__name__,
)
async def execute(self):
try:
self.log.info("Executing strategy C...")
await asyncio.sleep(1)
raise RuntimeError(self.ERROR_MESSAGE)
self.log.info("Strategy C execution finished.") # This line will never be reached
except Exception as e:
self.log.exception("An error occurred during strategyC execution", exception=str(e))
if __name__ == "__main__":
from logger_config import configure_logger
import uuid
configure_logger()
run_id = str(uuid.uuid4())
strategy_a = StrategyA(run_id)
asyncio.run(strategy_a.execute())
import pytest
from strategies import StrategyA
from strategies import StrategyB
from strategies import StrategyC
@pytest.fixture()
def run_id():
return "test_run"
@pytest.mark.asyncio()
async def test_strategyA(capsys, run_id):
strategy_a = StrategyA(run_id)
await strategy_a.execute()
captured = capsys.readouterr()
assert "Executing strategy A..." in captured.out
assert "Strategy A execution finished." in captured.out
@pytest.mark.asyncio()
async def test_strategyB(capsys, run_id):
strategy_b = StrategyB(run_id)
await strategy_b.execute()
captured = capsys.readouterr()
assert "Executing strategy B..." in captured.out
assert "Strategy B execution finished." in captured.out
@pytest.mark.asyncio()
async def test_strategyC(capsys, run_id):
strategy_c = StrategyC(run_id)
await strategy_c.execute()
captured = capsys.readouterr()
assert "Executing strategy C..." in captured.out
assert "An error occurred during strategyC execution" in captured.out
assert StrategyC.ERROR_MESSAGE in captured.out
実行結果
main.pyの実行結果
strategies.pyの実行結果
pytest の結果
実践コードの解説と補足
テストについて
Pytestにはloggerの出力をキャプチャする機能が備わっており、私はそれを利用してテストを書いている。(structlog自体にもtest関連の機能があるが私は使ってない)
captured = capsys.readouterr()
assert "Executing strategy C..." in captured.out
inspect と CallsiteParameterAdder について
class StrategyB:
def __init__(self, run_id):
self.run_id = run_id
+ self.log = structlog.get_logger(__name__).bind(
run_id=self.run_id,
+ strategy_name=self.__class__.__name__,
)
async def execute(self):
self.log.info(
+ "Executing strategy B...", function_name=inspect.currentframe().f_code.co_name
)
await asyncio.sleep(3)
self.log.info(
+ "Strategy B execution finished.", function_name=inspect.currentframe().f_code.co_name
)
上の部分ではロギングの処理に __name__ と __class__.__name__ と inspect.currentframe().f_code.co_name の値を追加している。
それぞれは実行時に strategies (あるいは __main__ )、 StrategyB 、 excecute として評価される。
structlog.processors.CallsiteParameter でも似た機能は実装できるが、私は使用していない。
structlogのasync機能について
私は使用していないが、structlogにはasync用の機能が存在する。
使い方は簡単で、logの実行メソッドの冒頭に a をつけるだけである。
しかし、私の実行例の中には非同期の処理は含まれてはいるが、structlogを非同期処理する操作は含んでいない。[6]
ログの参照について 追記(2024/3/7)
日本語でlogfmtへの対応の話があり、興味深かったので追記する。
私はローカル等、小さな範囲で行った実験等の結果はJSONで保存し、それをパースしてログを見ている。
pamburus/hl: A log viewer that translates JSON logs into human-readable representation.
各種クラウドサービスのログ管理SaaSは固有のフォーマットを指定してくることがある。その場合でも公式ドキュメントを見たり、それに対応するために作られた各種ライブラリの実装にあるログフォーマットのhandlerをstructlogに渡す。structlogのドキュメントに記載されているELK対応に従うなどして、簡単にstructlogは対応することができる。
私はlogfmtを使いたいと思ったことはないが、それも含めオプションは幅広く用意されていて柔軟に選択できるし、カスタマイズも容易なところがstructlogの便利なところだ。
github.com: orsinium-labs/gcp-log: Python library to format logs as GCP-compatible JSON.
github.com: kitchenita/python-logger-cloudwatch-structlog: Python library that allows logging in an AWS CloudWatch
gist.github.com: Class for logging structlog entries to a SQLite database
structlogの今後と寄付
キラーツール
私はstructlogをPythonのキラーツールだと考えている。
具体的には、Pythonユーザーのほどんどが書いたことがあるであろう print() は log.info() に置き代わるべきだと考えている。
大きなコードを書くときは当然、小さなスクリプトを書く場合であってもstructlogを使用できるなら log.info() の方が print() よりも生産性が高い。
Pythonを書くときはstructlogを使うべきだ。
不足しているもの
しかし、structlogの欠点はドキュメントの不足とテストの不足にある。
structlogの実装と機能は複雑であり、また対応しなくてはならないユースケース、エッジケースの組み合わせは無限に存在する。
しかも、それは読む地獄として悪名高いPythonのloggerAPIのラッパーなのである。
それをhynekがOSSとして一人でメンテナンスをし続けている。彼の凄さは筆舌に尽くしがたい。[7]
structlogはproduction-readyを謳っており、普通の動作の範疇ではバグはほぼなく安定している。
私は実装を読みながらエッジケースで遊んでいたので、幾つかコミットを投げようと思ったが、修正には全体の仕様を深く理解する必要があり、フルタイムのOSSメンテナーでないと厳しいように感じた。
コードで貢献できない私はGithub Sponsorをした。[8]
Sponsor @hynek on GitHub Sponsors
参考情報
-
この記事を書くために書いたコード
ZoraZen/Structlog_in_Action -
公式ドキュメント
structlog 23.1.0 documentation -
GitHub Sponsersのページ
Sponsor @hynek on GitHub Sponsors
Discussion