s-canvas/harness/crash_logger.py

"""S-CANVAS 크래시/예외 로깅 — sys.excepthook + faulthandler + 회전 파일 핸들러.

사용:
    from harness.crash_logger import install_crash_handlers
    install_crash_handlers()  # main() 진입점에서 한 번 호출

동작:
- 미처리 예외: traceback을 logs/scanvas.log + logs/crash_<ts>.txt에 저장
- C-level 크래시(segfault 등): faulthandler가 stderr + logs/faulthandler.log로 trace
- 메인 thread 외 thread 예외도 캡처 (threading.excepthook)
- 기존 stdout/stderr 출력은 유지 (사용자 인지 가능)
"""
from __future__ import annotations

import datetime as _dt
import faulthandler
import logging
import logging.handlers
import sys
import threading
import traceback
from pathlib import Path

# 상수
_LOG_DIR = Path(__file__).resolve().parent.parent / "logs"
_MAIN_LOG = _LOG_DIR / "scanvas.log"
_FAULT_LOG = _LOG_DIR / "faulthandler.log"
_MAX_BYTES = 5 * 1024 * 1024  # 5MB per file
_BACKUP_COUNT = 5             # rotate up to scanvas.log.5

_installed = False
_crash_logger: logging.Logger | None = None


def _ensure_log_dir() -> None:
    _LOG_DIR.mkdir(parents=True, exist_ok=True)


def _build_logger() -> logging.Logger:
    logger = logging.getLogger("scanvas.crash")
    logger.setLevel(logging.DEBUG)

    # 회전 파일 핸들러 (중복 install 방지: 핸들러 이미 있으면 재사용)
    if not any(isinstance(h, logging.handlers.RotatingFileHandler)
               for h in logger.handlers):
        handler = logging.handlers.RotatingFileHandler(
            _MAIN_LOG, maxBytes=_MAX_BYTES, backupCount=_BACKUP_COUNT,
            encoding="utf-8",
        )
        fmt = logging.Formatter(
            "%(asctime)s | %(levelname)s | %(threadName)s | %(message)s",
            datefmt="%Y-%m-%dT%H:%M:%S",
        )
        handler.setFormatter(fmt)
        logger.addHandler(handler)

    return logger


def _dump_crash_artifact(prefix: str, body: str) -> Path:
    """크래시별 별도 파일에도 전체 traceback 보관 (회전 로그와 별개)."""
    ts = _dt.datetime.now().strftime("%Y%m%dT%H%M%S")
    path = _LOG_DIR / f"{prefix}_{ts}.txt"
    path.write_text(body, encoding="utf-8")
    return path


def _excepthook(exc_type, exc_value, exc_tb):
    """sys.excepthook — 메인 thread 미처리 예외."""
    if exc_type is KeyboardInterrupt:
        # Ctrl+C는 정상 흐름으로 처리 (기본 동작 유지)
        sys.__excepthook__(exc_type, exc_value, exc_tb)
        return
    body = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
    if _crash_logger:
        _crash_logger.critical("UNCAUGHT EXCEPTION (main thread):\n%s", body)
    artifact = _dump_crash_artifact("crash", body)
    # 기존 동작 유지 — stderr에도 출력
    sys.__excepthook__(exc_type, exc_value, exc_tb)
    print(f"\n[crash_logger] 크래시 덤프 → {artifact}", file=sys.stderr)


def _thread_excepthook(args: threading.ExceptHookArgs) -> None:
    """threading.excepthook — 워커 thread 미처리 예외 (Py3.8+)."""
    if args.exc_type is SystemExit:
        return
    body = "".join(traceback.format_exception(
        args.exc_type, args.exc_value, args.exc_traceback))
    thread_name = args.thread.name if args.thread else "<unknown>"
    if _crash_logger:
        _crash_logger.error(
            "UNCAUGHT EXCEPTION (thread %s):\n%s", thread_name, body)
    _dump_crash_artifact(f"crash_thread_{thread_name}", body)


def install_crash_handlers(also_install_faulthandler: bool = True) -> Path:
    """크래시 핸들러를 한 번만 install. 재호출은 no-op.

    Returns:
        로그 디렉토리 경로 (사용자에게 보여줄 수 있음).
    """
    global _installed, _crash_logger  # noqa: PLW0603 (singleton install pattern)
    if _installed:
        return _LOG_DIR

    _ensure_log_dir()
    _crash_logger = _build_logger()

    # Python 예외 훅
    sys.excepthook = _excepthook
    threading.excepthook = _thread_excepthook

    # C-level 크래시 (segfault 등) — 별도 파일에 기록
    if also_install_faulthandler:
        # 기존 stderr 출력은 그대로 남기고, 추가로 파일에도.
        # NOTE: faulthandler는 long-lived file handle이 필요 (크래시 시점에 쓰기 위해)
        #       — context manager로 닫으면 안 됨.
        try:
            fh = open(_FAULT_LOG, "a", encoding="utf-8")  # noqa: SIM115 (faulthandler needs long-lived fh)
            faulthandler.enable(file=fh, all_threads=True)
        except OSError as e:
            # 파일 핸들 못 열면 stderr만으로 폴백
            faulthandler.enable(all_threads=True)
            _crash_logger.warning("faulthandler 파일 모드 실패 (%s) — stderr만 사용", e)

    _installed = True
    _crash_logger.info("crash_logger 설치 완료 (logs=%s)", _LOG_DIR)
    return _LOG_DIR


def get_logger() -> logging.Logger:
    """app.log() 같은 일반 로깅도 같은 회전 파일에 쓰고 싶을 때 사용."""
    if _crash_logger is None:
        # auto-install (설치 안 한 경우 안전한 폴백)
        install_crash_handlers()
    return _crash_logger  # type: ignore[return-value]