구조화된 로깅

구조화된 로그 기록이란

import logging, json

def log_json(level, msg, **kw):
    record = {"level": level, "msg": msg, **kw}
    print(json.dumps(record))

log_json("INFO", "Request received", path="/api/users", method="GET")

사용자 지정 JSON 형식 지정기

import logging, json

class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "time": self.formatTime(record),
            "level": record.levelname,
            "name": record.name,
            "msg": record.getMessage(),
        })

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.getLogger().addHandler(handler)

python-json-logger

# pip install python-json-logger
import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(jsonlogger.JsonFormatter())
logger.addHandler(handler)
logger.info("User created", extra={"user_id": 42})

structlog

# pip install structlog
import structlog

log = structlog.get_logger()
log = log.bind(request_id="abc123")
log.info("payment_processed", amount=49.99, currency="USD")

structlog로 컨텍스트 바인딩

import structlog

log = structlog.get_logger().bind(
    service="auth",
    env="prod"
)
log.warning("login_failed", user="alice", reason="bad_password")
# {service:auth, env:prod, user:alice, reason:bad_password}

타임스탬프와 호출자 정보 추가

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.CallsiteParameterAdder([structlog.processors.CallsiteParameter.FILENAME]),
        structlog.processors.JSONRenderer(),
    ]
)

웹 앱에서 요청 ID 기록

import structlog

def middleware(request, call_next):
    req_id = request.headers.get("X-Request-Id", generate_id())
    log = structlog.get_logger().bind(request_id=req_id)
    log.info("request_started", path=request.url.path)
    response = call_next(request)
    log.info("request_finished", status=response.status_code)
    return response

ECS 로그 형식

# pip install ecs-logging
import logging
import ecs_logging

logger = logging.getLogger("app")
handler = logging.StreamHandler()
handler.setFormatter(ecs_logging.StdlibFormatter())
logger.addHandler(handler)
logger.info("Event logged in ECS format")

로그 집계 시스템

# Loki query example:
# {app="myservice"} | json | level="ERROR"

# CloudWatch Insights:
# fields @timestamp, level, msg
# | filter level = "ERROR"
# | sort @timestamp desc

로그에서 PII 피하기

import re

def mask_pii(_, __, event_dict):
    msg = event_dict.get("event", "")
    event_dict["event"] = re.sub(
        r"[\w.+-]+@[\w-]+\.[\w.]+", "***@***.***", msg
    )
    return event_dict

구조화된 로그와 비구조화된 로그의 절충점

import structlog, os

if os.getenv("ENV") == "production":
    renderer = structlog.processors.JSONRenderer()
else:
    renderer = structlog.dev.ConsoleRenderer()

structlog.configure(processors=[renderer])

빠른 확인

복습