普段Pythonを主に使っているのですが、クラウドサービスでログ収集・分析基盤を整備する際に構造的データの方が何かと便利ですよね。しかし、Pythonの標準ライブラリのLoggingのノーマルなFormatterを使用すると一回構造的データに変換しなければなりません。コンバートするためだけにもう一個アプリケーションを作成するのは野暮です。
そこで、JSONでログを出力してくれるFormatterを開発してみました。

環境

OS (動作保証済み)
- MacOS Catalina
- Ubuntu 18.04
言語
- Python ^3.7
パッケージマネージャー
- Poetry
ライブラリ
- 標準ライブラリのみ
開発用ライブラリ
- Black
- Pytest
- Flake8

成果物

開発したFormatterはhomoluctus/json-pyformatterにあります。
PyPIにも公開してあります -> json-pyformatter 0.1.0

使用方法

1. インストール

pip install json-pyformatter

2. 例

出力できるフィールドは標準ライブラリLoggingのlogrecord-attributesです。
開発したFormatterが出力するフィールドのデフォルトではasctime、levelname, messageです。

import logging
from json_pyformmatter import JsonFormatter

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
fields = ('levelname', 'filename', 'message')
formatter = JsonFormatter(fields=fields)
handler.setFormatter(formatter)
logger.addHandler(hander)

logger.info('hello')

これを実行すると以下のようにJSONのログが出力されます。

{"levelname": "INFO", "filename": "test_formatter.py", "message": "hello"}

また、JsonFormatterの引数にindent=2を指定してあげるとより見やすくなります。

{
  "levelname": "INFO",
  "filename": "test_formatter.py",
  "message": "hello"
}

もちろんtracebackを出力することもできます。1行ですと見にくいので配列にしています。

{
  'asctime': '2019-12-01 13:58:34',
  'levelname': 'ERROR',
  'message': 'error occurred !!',
  'traceback': [
    'Traceback (most rec...ll last):',
    'File "/example/test..._exc_info',
    'raise TypeError(message)',
    'TypeError: error occurred !!'
  ]
}

ソースコード解説

ここからはソースコードの解説をしていきます。

全ソースコード

とりあえず全ソースコードを書いておきます。

import json
from collections import OrderedDict
from logging import Formatter


class JsonFormatter(Formatter):
    default_fields = ('asctime', 'levelname', 'message')

    def __init__(self, fields=None, datefmt=None, indent=None):
        """
        Args:
            fields (tuple, list)
            datefmt (str)
            indent (str, int)
        """

        self.fields = (
            self.get_or_none(fields, (list, tuple)) or self.default_fields
        )
        # default time format is %Y-%m-%d %H:%M:%S
        self.datefmt = (
            self.get_or_none(datefmt, str) or self.default_time_format
        )
        self._indent = self.get_or_none(indent, (str, int))

    def get_or_none(self, target, types):
        """Check whether target value is expected type.
        If target type does not match expected type, returns None.

        Args:
            target (any)
            types (class, tuple)

        Returns:
            target or None
        """

        if isinstance(target, types):
            return target
        return None

    def getMessage(self, record):
        if isinstance(record.msg, (list, tuple, dict)):
            return record.msg
        return record.getMessage()

    def _format_json(self, record):
        return json.dumps(record, ensure_ascii=False, indent=self._indent)

    def _format(self, record):
        log = OrderedDict()

        try:
            for field in self.fields:
                log[field] = getattr(record, field)
            return log
        except AttributeError as err:
            raise ValueError(f'Formatting field not found in log record {err}')

    def format(self, record):
        record.message = self.getMessage(record)
        record.asctime = self.formatTime(record, self.datefmt)
        formatted_record = self._format(record)
        if record.exc_info:
            if not record.exc_text:
                record.exc_text = self.formatException(record.exc_info)
        if record.exc_text:
            formatted_record['traceback'] = [
                msg.strip() for msg in record.exc_text.strip().split('\n')
            ]
        if record.stack_info:
            formatted_record['stack'] = record.stack_info.strip()

        return self._format_json(formatted_record)

get_or_none

インスタンス作成時に使用するメソッドです。
期待する型でなければデフォルトの値を使用したり、そのままNoneを代入するために使います。

def get_or_none(self, target, types):
    """Check whether target value is expected type.
    If target type does not match expected type, returns None.

    Args:
       target (any)
       types (class, tuple)

    Returns:
        target or None
    """

    if isinstance(target, types):
        return target
    return None

getMessage

logger.info()の引数のメッセージを取得するためのメソッドです。このgetMessageの返り値がJSONのmessageフィールドにセットされます。
record.msgのインスタンスタイプが(list, tuple, dict)のいずれかであればそのまま返し、それ以外であればrecordインスタンスのgetMessageメソッドの返り値を返します。そうすることで(list, tuple, dict)のいずれかが渡された時にJSONの配列・オブジェクトとしてログを出力できます。

def getMessage(self, record):
    if isinstance(record.msg, (list, tuple, dict)):
        return record.msg
    return record.getMessage()

_format_json

引数のrecordをJSONとしてdumpするメソッドです。

def _format_json(self, record):
    return json.dumps(record, ensure_ascii=False, indent=self._indent)

_format

recordをPythonの辞書に変換するメソッドです。
必ずしもユーザーが指定したフィールドがrecordのアトリビュートに存在しているわけではないのでtry-exceptしてあります。返り値のlogはOrderedDictにして順序を保証しています。

def _format(self, record):
    log = OrderedDict()

    try:
        for field in self.fields:
            log[field] = getattr(record, field)
        return log
    except AttributeError as err:
        raise ValueError(f'Formatting field not found in log record {err}')

format

このメソッドがlogging.Handlerから呼び出されます。Formatterを自作して親クラス(logging.Formatter)のformatを使うことはないと思いますので、このメソッドは必ず上書きします。
tracebackを配列として出力すために、record.exc_textを整形しています。余分な空白などを削除したあとに改行でスプリットして配列にしています。
最後にPythonのdictをJSONとしてダンプしています。

def format(self, record):
    record.message = self.getMessage(record)
    record.asctime = self.formatTime(record, self.datefmt)
    formatted_record = self._format(record)
    if record.exc_info:
        if not record.exc_text:
            record.exc_text = self.formatException(record.exc_info)
    if record.exc_text:
        formatted_record['traceback'] = [
            msg.strip() for msg in record.exc_text.strip().split('\n')
        ]
    if record.stack_info:
        formatted_record['stack'] = record.stack_info.strip()

    return self._format_json(formatted_record)

PythonのログをJSONで出力するFormatterを開発してみた

環境

成果物