🗂

Pythonのstructlogを使いこなす

2023/06/06に公開

記事の内容

  • 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の革新性は、従来のロギングが抱えていた「高機能なログの実現」と「使いやすさ」のジレンマを克服した点にある。
従来の構造化ロギングは各要素への詳細なパラメーター設定を必要とし、それはコードの複雑化を招いた。ログの内容を作り込むほどコードの複雑性はべき乗に膨れ上がっていき、扱い易さは損なわれた。しかし、structlogは直感的で操作性のまま、ユーザー側の高度な要求を簡単に実現できる。
このノベーションは作者hynekの卓越したAPI設計と高度な実装能力によって実現された。その他に実行速度のメリットもある。全体を評するなら作者の主張通り「シンプル、パワフル、ファスト」の一石三鳥の評価が妥当だろう。

他ツールとの比較

  • 著者の主観で他のロガーツール(loggerモジュール[2]logru[3])とstructlogの比較をする

1. Pythonのloggerモジュール vs structlog

- loggerモジュールの利点
  - Pythonのビルトインであるため、ライブラリを導入する必要がない
- loggerモジュールの欠点
  - 設計がJavaのライブラリの移植でAPIのデザインがPythonと一貫性がない
  - そのためドキュメントを解読するのに数日かかる
  - 正しく使えるようになるには一週間以上、長ければ年単位の時間を要する

2. logru vs structlog

- logruの利点
  - 導入方法がシンプルで最初のlogを成功させるまでは最速
  - 対してstructlogは使い始めにドキュメントを一時間くらい読む必要がある
- logruの欠点
  - loggerモジュールの薄いラッパーであり深入りするなら、loggerモジュールと変わらない
  - structlogはloggerモジュールのラッパーとして扱え、
    その場合でもシンプルな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
"""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のスクリプトを以下のように作成し実行する。

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形式で保存すること

達成しないこと

  1. クラウドサービスのログ管理ツールとの連携
  2. ELKスタックなどの中央ログ管理ツールとの連携
  3. ロギング処理の高速化・並列処理の実装
  4. ログの圧縮保存

この記事ではこれらの要素について言及しないが公式ドキュメントには関連する記載がある。[5]

詳細解説

logger_config.py
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_stdouthandler_file をstructlog上で用意し、それをPythonのloggerモジュールに渡す。

そのhandler_stdout (ターミンル出力のハンドラ)には structlog.dev.ConsoleRendererhandler_file (ファイル出力のハンドラ)には structlog.processors.JSONRenderer をセットする。
ConsoleRenderer はターミナル上で視認性の高いログを表示する。
JSONRenderer はログをapplication.logjson形式で保存する。

拡張

以下のように引数を作って使うことも多い。
logging.LEVELの実態は整数であるが、私はその対応関係を覚えていない、かつ、数値のためだけにloggingをimportしたくないので文字列で呼べるようにしている。なおJavaと違い、CRITICALとFATALは同列である。

logger_config.py
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)

テンプレートの使用例

以下はテンプレートを使用した構造化ロギングの例である。
解説は後述する。

main.py
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())
strategies.py
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())
test_strategies.py
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の実行結果
main.pyの実行結果

strategies.pyの実行結果
strategies.pyの実行結果

テストの結果
pytest の結果

実践コードの解説と補足

テストについて

Pytestにはloggerの出力をキャプチャする機能が備わっており、私はそれを利用してテストを書いている。(structlog自体にもtest関連の機能があるが私は使ってない)

test_strategies.py
captured = capsys.readouterr()
assert "Executing strategy C..." in captured.out

inspectCallsiteParameterAdder について

strategies.py
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__ )、 StrategyBexcecute として評価される。
structlog.processors.CallsiteParameter でも似た機能は実装できるが、私は使用していない。

structlogのasync機能について

私は使用していないが、structlogにはasync用の機能が存在する。
使い方は簡単で、logの実行メソッドの冒頭に a をつけるだけである。
しかし、私の実行例の中には非同期の処理は含まれてはいるが、structlogを非同期処理する操作は含んでいない。[6]

ログの参照について 追記(2024/3/7)

日本語でlogfmtへの対応の話があり、興味深かったので追記する。
https://methane.hatenablog.jp/entry/2024/03/04/構造化ログのフォーマット_logfmt_vs_JSON_lines

私はローカル等、小さな範囲で行った実験等の結果は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
スポンサー画面

参考情報

脚注
  1. メンテナンスも Hynek が行っている ↩︎

  2. Pythonに含まれているモジュールとしてのlogger ↩︎

  3. Delgan/loguru: Python logging made (stupidly) simple ↩︎

  4. チュートリアル、あるにはあるが、かなり上級者向き
    Getting Started - structlog 23.1.0 documentation ↩︎

  5. Logging Best Practices - structlog 23.1.0 documentation ↩︎

  6. 今後、パフォーマンスを追求する必要が生じれば動作検証をして使用すると思うが今はモチベーションがない ↩︎

  7. 昔から凄い人なのだけれども ↩︎

  8. この記事はGitHub Sponsorのステマです ↩︎

GitHubで編集を提案

Discussion