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のloggerモジュール上でも構造化ログは実現できるが、structlogは最初から構造化ログに特化したAPIが用意されているという利点がある。
この「構造化ログに特化した仕組み」とはインターフェース面においては、ユーザーは辞書型データ構造を用いて、キーと値のペアでログを簡単に生成できる仕組みである。
この「キーと値のペアでログを作る」というのは、ライフタイム管理などの多くのパラメーターを生み、操作を複雑にし、混乱を招きそうな仕様である。だが実際にはstructlogのAPIは直感的に混乱なく操作できて、かつ高度な要求、細かな調整にも対応できる。ここで矛盾しているが、それは作者の技術力・設計能力・イノベーションによって解決されている。
結果として、structlogを使うことで、ユーザーはPythonのloggerよりも直感的、かつ効率的に可読性と一貫性があり保全性・保守性に長けたlogを準備できる。
他ツールとの比較
1. Pythonのloggerモジュール vs structlog
- loggerモジュールの利点
- Pythonのビルトインであるため、ライブラリを導入する必要がない。
- loggerモジュールの欠点
- ドキュメントを解読するのに数日かかる
- 正しく使うのにさらに数日かかる
2. logru vs structlog
- logruの利点
- 導入方法がシンプルでlogを成功させるまでは最速。
- structlogは使い始めにドキュメントを一時間くらい読む必要がある。
- logruの欠点
- logruは現状ではloggerモジュールの薄いラッパーなので、深入りするなら手間はloggerモジュールと変わらない。
- structlogは設定を変更するのが段違いに簡単。
- structlogはloggerモジュールのラッパーとして使うこともできる。
- その場合でも設定はstructlogのAPIを通じて記載することができloggerモジュールの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: logの結果を
aplication.logにjson形式で保存すること
達成しないこと
- クラウドサービスのログ管理ツールとの連携
- ELKスタックなどの中央ログ管理ツールとの連携
- logging処理の高速化・並列処理の実装
- logの圧縮保存
この記事ではこれらの要素について言及しないが公式ドキュメントには関連する記載がある。^[Logging Best Practices - structlog 23.1.0 documentation
詳細解説
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 をセットする。
JSONRenderer はログをapplication.logにjson形式で保存する。
ConsoleRenderer はターミナル上でログを表示する。
その際、ただ表示するのみならず、PythonのCUI表示を改善するライブラリである colorama[5]と better-exceptions 、 Rich のパッケージがインストールされていれば、それらが使用され可読性が向上する。
拡張
以下のように引数を作って使うことも多い。
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の出力をキャプチャする機能が備わっており、私はそれを利用してテストを書いている。
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.
各種クラウドサービスはそれぞれに専用のlogのフォーマットを指定してくることがある。その場合でも公式ドキュメントを見たり、それに対応するために作られた各種ライブラリの実装にあるlogフォーマットの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