Pythonのログ出力を自分でJSON形式にする

Pythonのログ出力は、デフォルトでは1つのログが複数行に渡って出力される。これはターミナル上で手動で実行する場合は見易いが、ログ収集のシステムと連携してしようとすると各行が1つのログとして扱われてしまう。これはログの集計の結果にも影響を与えてしまうし、なにより検索しにくくなってしまう。そのため、PythonのloggerのフォーマッターをJSON形式用のフォーマッターに差し替えて使う。structlogやpython-json-formatterは、そういったライブラリだ。

ただJSON形式のログを出力するだけの為に、何でサードパーティのライブラリに依存しないといけないのだろうか。Pythonにはjsonモジュールもloggingモジュールもあるのに、ログ変換処理のためだけに外部の何かに依存したくない。

そこで、この変換処理は自分で簡易なものを実装する事にした。

main.py

import json
import logging
import traceback


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger_name": record.name,
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
            "thread": record.threadName,
            "process": record.process,
            "traceback": "".join(traceback.format_tb(record.exc_info[2]))
            if record.exc_info
            else None,
        }
        return json.dumps(log_data, ensure_ascii=False)


handler = logging.StreamHandler()
formatter = JSONFormatter()
handler.setFormatter(formatter)

logging.basicConfig(level=logging.INFO, handlers=[handler])


logger = logging.getLogger("custom-logger")
logger.info("Ok")


def testing():
    raise ValueError("日本語例外")


try:
    testing()
except:
    logger.exception("？Oops!!", exc_info=True)

python main.py 2>&1

{"timestamp": "2023-08-04 15:11:45,652", "level": "INFO", "message": "Ok", "logger_name": "custom-logger", "module": "main", "function": "<module>", "line": 33, "thread": "MainThread", "process": 55413, "traceback": null}
{"timestamp": "2023-08-04 15:11:45,652", "level": "ERROR", "message": "？Oops!!", "logger_name": "custom-logger", "module": "main", "function": "<module>", "line": 43, "thread": "MainThread", "process": 55413, "traceback": "  File \"/opt/ng/symdon/pages/posts/1691127585/main.py\", line 41, in <module>\n    testing()\n  File \"/opt/ng/symdon/pages/posts/1691127585/main.py\", line 37, in testing\n    raise ValueError(\"日本語例外\")\n"}

AWS CloudWatch Logs だとログのレコードを正しく認識できなくて、1つのレコードに複数のログが混在してしまう状態になる事があった。この処理では不十分だと思う。structlogではどのように実装しているのか、時間を見つけて調査してみたい。

class CustomLoggingAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        if "extra" not in kwargs:
            kwargs["extra"] = {}

        kwargs["extra"]["custom_value"] = "Hello world!"
        return msg, kwargs

logger = logging.getLogger(__name__)
log_adapter = CustomLoggingAdapter(logger)