logger.py

import logging
from typing import Literal, Optional

import structlog
from structlog_sentry import SentryProcessor


def create_logger(
    mode: Literal["json", "console"],
    instance_name: Optional[str] = None,
    container_id: Optional[str] = None,
    log_level: int = logging.INFO,
) -> structlog.stdlib.BoundLogger:
    match mode:
        case "json":
            renderer: structlog.typing.Processor = structlog.processors.JSONRenderer()
        case "console":
            renderer = structlog.dev.ConsoleRenderer()
        case _:
            raise ValueError(f"Invalid mode: {mode}")
    structlog.configure(
        processors=[
            structlog.processors.add_log_level,
            structlog.contextvars.merge_contextvars,
            structlog.processors.StackInfoRenderer(),
            structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
            SentryProcessor(event_level=logging.ERROR, tag_keys="__all__"),
            structlog.processors.format_exc_info,
            renderer,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(log_level),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=False,
    )
    my_logger = structlog.get_logger()

    if container_id is not None:
        my_logger = my_logger.bind(container_id=container_id)
    if instance_name is not None:
        my_logger = my_logger.bind(instance_name=instance_name)

    return my_logger

mode で構造化jsonか人間に優しい形式かを設定できます。

2024-04-17 13:47:10 [info     ] worker started                
2024-04-17 13:47:10 [info     ] start queue watch             
2024-04-17 13:47:10 [info     ] start queue watch             
2024-04-17 13:47:11 [info     ] health check failed. ignore... class_name=QueueProcessor
2024-04-17 13:47:11 [info     ] health check failed. ignore... class_name=QueueProcessor

{"event": "worker started", "level": "info", "timestamp": "2024-04-17 13:48:49"}
{"event": "start queue watch", "level": "info", "timestamp": "2024-04-17 13:48:49"}
{"event": "start queue watch", "level": "info", "timestamp": "2024-04-17 13:48:49"}
{"class_name": "QueueProcessor", "event": "health check failed. ignore...", "level": "info", "timestamp": "2024-04-17 13:48:50"}
{"class_name": "QueueProcessor", "event": "health check failed. ignore...", "level": "info", "timestamp": "2024-04-17 13:48:50"}

構造化ログのテンプレート

level

logger = create_logger()

logger.info("event", context="context value")

sentry

ERROR以上のログを自動的にsentryに送信します。

logger = create_logger()

logger.error("failed to process")

...

except Exception:
  logger.error("error", exc_info=True) # exc_info=Trueでログにスタックトレースが乗ります
  logger.exception("exception") # exceptionはerror exc_info=Trueのシノニム

context

たとえばジョブIDなどのコンテキストを含めることができます。
コンテキストはバインドすることができます。一時的に同じコンテキストを含めたい場合に便利です。

logger.info("event", context="dev")

log = logger.bind(context="dev")
log.info("event") # 上のlogger.infoと同等の挙動
log.error("error") # ここでもcontextが入る