https://nealle-dev.hatenablog.com/entry/2025/04/04/170711

こんにちは。SREチームの高 (@nogtk)です。ゼノブレイドXリメイクで惑星ミラの探索に勤しんでいる今日この頃です。

直近行った取り組みとして、アプリケーションログの構造化を行い、ログの検索性の向上を行いました。この記事では実際の実装も交えつつ実施した内容についてご紹介したいと思います。

構造化ロギングによって解決したい課題

Park Direct のバックエンドアプリケーションは Django で作られ、ログの出力先としては Datadog Logs を利用しています。このアプリケーションログは、長らくデフォルトである非構造化形式でログが出力されており、以下のような課題を抱えていました。

タイムスタンプやログレベル、トレースバックなどの各要素について、Grokパーサなどを駆使し Datadog のログパイプライン処理で抽出を行なっていたが、パターン網羅に限界があり、うまくパースできていないログが存在している
ログとして出力する項目が標準化されておらず、呼び出し元の様々なコンテキストで多様なログを出力しており、エラーログの監視やトラブルシューティングが効率よく行えない

これらを踏まえ

ログとして出力する項目を標準化すること
その項目を構造化ログとして出力すること

以上2点をゴールとして設定し、進めていくこととなりました。

最終的にできるようになったことの紹介

まずはこの取り組みによってどう検索性が向上したかについてご紹介したいと思います。

リクエスト内容について、user_id/request_id など各タグで検索できる

非同期ジョブのログについて、task_id(ジョブごとに振られるユニークID) で検索できる

いかがでしょうか？手前味噌ではありますが、検索性がかなり向上したと言えると思います。では、これらを実現するにあたってどう進めていったのかご紹介していきたいと思います。

ログ項目の決定

ログの項目については、以下の観点を踏まえながらアプリケーション開発者も巻き込み決定していきました。

現状どのようにログを使ってトラブルシューティングを行っているのか、についてのヒアリング結果
Datadog Standard Attributes で言及されているタグ
Datadog Logs と APM の連携に必要なタグ

どういう項目を標準とするかは悩んだ部分ですが、運用上必要となった場合は後から追加できるという前提で、まずは決めることを優先させ、以下のように16項目を定義しました。

1. level ※ログレベル
2. logger
3. timestamp
4. user_id
5. message
6. request_id
7. request_method
8. request_path
9. referer
10. user_agent
11. dd.version
12. dd.env
13. dd.service
14. exception_class ※エラーログの場合のみ
15. stacktrace ※エラーログの場合のみ
16. task_id ※非同期ジョブの場合のみ

ログライブラリの選定

ログの項目が決まったので、後はこれらを出力できるように実装を進めていく形になります。手始めに構造化ロギング用にどのライブラリを使うかについて検討を行いました。候補としては https://docs.datadoghq.com/ja/logs/log_collection/python/ で言及があるものに加えて、その他いくつかピックアップしました。

https://pypi.org/project/JSON-log-formatter/
https://pypi.org/project/python-json-logger/
https://pypi.org/project/django-datadog-logger/
https://pypi.org/project/loguru/
https://pypi.org/project/structlog/
ライブラリを使わない自前実装

リポジトリのスター数、開発チームからのアドバイス、Django 用の拡張ライブラリがある点などを考慮し、今回は structlog の Django 向けインテグレーションを含んだ django-structlog を採用する形で着地をしました。

実装について

dictConfig クラス

Park Direct の Django アプリケーションにおけるログの設定には、デフォルトと同じ dictConfig クラスを用いています。詳細は公式ドキュメントに記載があるので詳細は割愛しますが、以下のようなフォーマットになっています。

{
    'formatters': {
      'json': {
            '()': 'structlog.stdlib.ProcessorFormatter',
            "processor": structlog.processors.JSONRenderer(ensure_ascii=False),
            "foreign_pre_chain": [...]
        },
    },
    'handlers': {
        'json': {
            'class': 'logging.StreamHandler',
            'formatter': 'json',
        },
    },
    'loggers': {
        'app': {
            'handlers': ['json'],
            'level': 'INFO',
            'propagate': True,
        },
    },
}

formatters, handlers, loggers という3つのフィールドがあり、logger からは handler を、handler からは formatter を数珠繋ぎのようにして指定し、ログの設定を行います。

今回の構造化ログの文脈においては、formatter と handler として json ログ用のものを定義し、既存の logger の handler を ‘json’ として置き換えることで、ログの呼び出し側の実装を変更することなく、この dictConfig フォーマットのログ設定で、一元管理することができます。

structlog の設定を dictConfig フォーマット上で指定する

structlog を導入し、JSONロギング用の設定を追加することで structlog.get_logger().info("hoge")と呼び出すと、構造化されたログを出力できます。しかし、これだと既存コードにおけるPython 標準のロガーを使っている呼び出し側の実装を変更する必要が出てきてしまいます。

そこで、dictConfig クラスに渡す設定に structlog の設定を組み込み、Python 標準の logging モジュール由来のログインターフェースは維持したまま構造化ログの設定を反映できるような工夫を施します。以下は json フォーマット用の設定について抜粋したものです。

'json': {
  '()': 'structlog.stdlib.ProcessorFormatter',
  'processor': structlog.processors.JSONRenderer(ensure_ascii=False),
  'foreign_pre_chain': [
    structlog.stdlib.add_log_level,
     ...(割愛)
   ]
}

structlog が提供する ProcessorFormatter を利用することで、logging モジュールから呼び出したロガーに対しても、structlog の設定を反映させることができます。JSON ロギングの設定自体は processor で行なっており、JSON ログのカスタマイズを foreign_pre_chain として指定をしています。

structlog のドキュメントには、これらの話の全体像が図解されています。

JSON ログのカスタマイズについて

foreign_pre_chain は、ProcessorFormatter のオプションとして用意されていて、様々なログのプロパティを追加したりログのプロパティ名を変更したりなどの設定が行えます。

例えば、上のサンプルで示した structlog.stdlib.add_log_level を呼び出すことで、その名の通りログレベルをログとして出力する項目に追加することができます。ドキュメントを参照しながら、必要なプロセッサを foreign_pre_chain へ追加していきます。

このプロセッサは、ライブラリ側で提供されているものだけではなく独自に作成することもできます。 processors としてドキュメントに記載がありますが、指定のインターフェースと指定の値を返すメソッドを実装することで、カスタムプロセッサを追加できます。Datadog 用のログ項目を追加するケースなどで、プロセッサを自作し同様に foreign_pre_chain に追加して実現をしています。

Datadog 用のタグを付与するカスタムプロセッサの実装例

# logger, method_name, event_dict の3引数を受け取る必要がある
def _datadog_injector(_logger, _method_name, event_dict):
        span = tracer.current_span()
        trace_id, span_id = (str((1 << 64) - 1 & span.trace_id), span.span_id) if span else (None, None)

        additional_datadog_tags = {
            'dd.trace_id': trace_id,
            'dd.span_id': span_id,
            'dd.env': ddtrace.config.env or "",
            'dd.service': ddtrace.config.service or "",
            'dd.version': ddtrace.config.version or "",
        }

        # event_dict に structlog に追加したタグが含まれているので、追加分をマージして返却
        return event_dict | additional_datadog_tags

'json': {
  '()': 'structlog.stdlib.ProcessorFormatter',
  'processor': structlog.processors.JSONRenderer(ensure_ascii=False),
  'foreign_pre_chain': [
    structlog.stdlib.add_log_level,
     ...(割愛)
     _datadog_injector,
   ]
}

HTTP のコンテキストをログに含める

出力されるログには、リクエストURLやリファラなど、HTTPリクエストのコンテキストを含める必要があります。structlog は structlog.contextvars というモジュールを提供しており、スレッドセーフなコンテキストに任意の値をバインドすることができます。

今回は HTTP に関連するデータをバインドしたかったので、リクエスト処理にフックを加えることができるDjango のミドルウェア内で、このバインド処理を呼び出します。

class LoggingContextBinderMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request: HttpRequest):
        structlog.contextvars.bind_contextvars(
            request_url=request.build_absolute_uri(),
            referrer=request.headers.get('Referer', None),
            // その他割愛
        )

        response = self.get_response(request)
        return response

bind_contextvars の引数にログとして出力したい項目を渡すことで、そのHTTPリクエスト内で出力されたログに、これらの情報を透過的に含めることができます。前述した foreign_pre_chain に structlog.contextvars.merge_contextvars を呼ぶことで、バインドしたデータも含めてログの出力に含めることができます。

Celery ログ

Park Direct では非同期タスク処理に Celery を採用しており、django-structlog は Celery もサポートをしています。

詳細はこちらに記載があるのですが、今回は bind_extra_task_metadata を使って、Celery の task_id をログ出力に含めるやり方を紹介したいと思います。

Celery の設定ファイルに、以下のようなレシーバーを定義します。

@receiver(signals.bind_extra_task_metadata)
def receiver_bind_extra_request_metadata(sender, signal, task=None, logger=None, **kwargs):
    structlog.contextvars.bind_contextvars(task_id=task.request.id if task else 0)

また Park Direct では、独自の実装としてタスクの実行時間やステータスの計測用に celery.app.task.Task の apply_async を拡張していた関係*1 で、この apply_async メソッド内で bind_extra_task_metadata シグナルを送出する処理も追加しました。

class CustomTask(Task):

    def apply_async(self, *args, **kwargs):
        kwargs['headers'] = {'sent_timestamp_ms': round(time.time() * 1000)}
        signals.bind_extra_task_metadata.send(sender=self.__class__, task=self)
        try:
            result = super().apply_async(*args, **kwargs)
            // 以下割愛

これによりワーカータスクの実行時に、task_id を structlog の context へとバインドする処理が呼ばれ、ワーカータスク内でのログ出力時において、 task_id の透過的な付与を実現することができました。

まとめ

今回は django-structlog を用いたDjango アプリケーションログの構造化について、実際の実装にも踏み込んで事例をご紹介致しました。参考になれば幸いです。

アプリケーション開発者からも、実際にログを使ったトラブルシューティングや調査時に便利になったという声をもらうことができました。引き続き、開発者にとってより Datadog が使いやすくなるように、アプリケーションの改善を進めていきたいと思います！

*1:こちらの経緯については非同期タスクのメトリクス収集術 - Nealle Developer's Blog を参照ください