galaxis-agent/agent/json_logging.py

56 lines
1.8 KiB
Python
Raw Normal View History

"""구조화 JSON 로깅.
LOG_FORMAT 환경변수로 json | text 선택 가능.
"""
from __future__ import annotations
import json
import logging
import traceback
from datetime import datetime, timezone
_BUILTIN_ATTRS = {
"args", "created", "exc_info", "exc_text", "filename", "funcName",
"levelname", "levelno", "lineno", "module", "msecs", "message", "msg",
"name", "pathname", "process", "processName", "relativeCreated",
"stack_info", "thread", "threadName", "taskName",
}
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
for key, value in record.__dict__.items():
if key not in _BUILTIN_ATTRS and not key.startswith("_"):
try:
json.dumps(value)
log_data[key] = value
except (TypeError, ValueError):
log_data[key] = str(value)
if record.exc_info and record.exc_info[0] is not None:
log_data["exception"] = "".join(traceback.format_exception(*record.exc_info))
return json.dumps(log_data, ensure_ascii=False)
def setup_logging(
log_format: str = "json",
level: int = logging.INFO,
logger: logging.Logger | None = None,
) -> None:
target = logger or logging.getLogger()
if log_format == "json":
formatter = JsonFormatter()
else:
formatter = logging.Formatter(
"%(asctime)s %(levelname)s %(name)s: %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
for handler in target.handlers:
handler.setFormatter(formatter)
target.setLevel(level)