普段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)