"""한 진단 process의 시작 경계만 기록한다. 기본 비활성, Runtime 정상 판정에는 사용하지 않는다. CONTEXTRA_START_DIAGNOSTICS_DIR은 사전 생성한 로컬 mode700 디렉터리여야 한다. bootstrap 최초 실행에서 10초 뒤 locals 없는 Python stack을 한 번 기록한다. stack FD는 faulthandler의 process-wide 소유권 때문에 process 종료까지 유지한다. 여러 Stream 동시 진단이나 운영 상시 관측용이 아니다. """ import faulthandler import json import os import stat import threading import time from contextlib import contextmanager from collections import deque STAGES = frozenset({"source", "profile", "config", "pipeline", "flow", "sdk_start", "sdk_wait", "source_prepare", "batch_capture", "source_property", "rtp_attach", "ingress_probe", "preprocess", "infer", "retrieve"}) ROLES = frozenset({"bootstrap", "det", "pose", "seg"}) _stack_lock = threading.Lock() _stack_file = None def setup_failure_summary(error): """Keep code locations, never exception text, paths, source lines or locals.""" chain = [] seen = set() while error is not None and id(error) not in seen and len(chain) < 3: seen.add(id(error)) frames = deque(maxlen=8) tb = error.__traceback__ for _ in range(64): if tb is None: break code = tb.tb_frame.f_code frames.append({"file": os.path.basename(code.co_filename)[:80], "function": code.co_name[:80], "line": tb.tb_lineno}) tb = tb.tb_next item = {"type": type(error).__name__[:80], "frames": list(frames)} if isinstance(error, OSError) and isinstance(error.errno, int): item["errno"] = error.errno chain.append(item) error = error.__cause__ or (None if error.__suppress_context__ else error.__context__) return chain class RoleDiagnostics: """진단 전용 고정 크기 snapshot. SDK/lifecycle lock이나 파일 I/O를 사용하지 않는다.""" EVENTS = frozenset({"bound", "start_enter", "start_returned", "stop_requested", "source_ready_stop", "source_wait_expired", "stop_unstarted", "stop_enter", "stop_returned", "stop_raised", "stop_no_pipeline", "stop_already_finished", "wait_enter", "wait_returned", "wait_raised", "worker_finally", "worker_cleanup_returned"}) COUNTERS = frozenset({"decoded_frame_received", "preprocess_output", "infer_output", "consume_entered", "dropped_before_infer"}) def __init__(self): self.enabled = bool(os.environ.get("CONTEXTRA_START_DIAGNOSTICS_DIR")) self.failure = None try: self.token = os.urandom(12).hex() if self.enabled else None except OSError: self.token = None self.enabled = False self._lock = threading.Lock() self._events = {} self._counters = {} self._stop_matches_started = None def mark(self, event, *, matches_started=None): if not self.enabled or event not in self.EVENTS: return try: now = time.monotonic_ns() with self._lock: self._events.setdefault(event, now) if event == "stop_enter": self._stop_matches_started = matches_started except Exception: self.failure = "OBSERVATION_FAILED" def count(self, stage): if not self.enabled or stage not in self.COUNTERS: return try: now = time.monotonic_ns() with self._lock: old = self._counters.get(stage) self._counters[stage] = { "count": min(old["count"] + 1, 2**63 - 1) if old else 1, "first_ns": old["first_ns"] if old else now, "last_ns": now, } except Exception: self.failure = "OBSERVATION_FAILED" def snapshot(self): if not self.enabled: return None with self._lock: return {"token": self.token, "diagnostic_error": self.failure, "events": dict(self._events), "counters": {key: dict(value) for key, value in self._counters.items()}, "stop_matches_started": self._stop_matches_started} class StartDiagnostics: def __init__(self, role): self.role = role self.fd = None self.failure = None self._seen = set() root = os.environ.get("CONTEXTRA_START_DIAGNOSTICS_DIR") if not root: return if role not in ROLES: self.failure = "INVALID_ROLE" return directory = None try: directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) info = os.fstat(directory) if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: raise OSError("invalid diagnostic directory") # pid/thread ID를 출력하지 않고 파일 충돌만 방지한다. name = os.urandom(12).hex() + ".jsonl" self.fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory) if role == "bootstrap": self._arm_stack(directory) except (OSError, ValueError): self.failure = "DIAGNOSTICS_OPEN_FAILED" finally: if directory is not None: os.close(directory) def _arm_stack(self, directory): global _stack_file with _stack_lock: if _stack_file is not None: return fd = None try: fd = os.open("python-stack.txt", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory) stream = os.fdopen(fd, "w") fd = None try: faulthandler.dump_traceback_later(10, repeat=False, file=stream, exit=False) except (OSError, RuntimeError): stream.close() raise _stack_file = stream except (OSError, RuntimeError): if fd is not None: os.close(fd) self.failure = "STACK_ARM_FAILED" def mark(self, stage, phase): if self.fd is None or stage not in STAGES or phase not in {"enter", "returned", "raised"}: return key = (stage, phase) if key in self._seen: return self._seen.add(key) payload = json.dumps({"stage": stage, "phase": phase, "thread_role": self.role, "monotonic_ns": time.monotonic_ns(), "diagnostic_error": self.failure}) try: data = (payload + "\n").encode("ascii") if os.write(self.fd, data) != len(data): raise OSError("short write") except OSError: self.failure = "DIAGNOSTICS_WRITE_FAILED" self.close() @contextmanager def span(self, stage): self.mark(stage, "enter") try: yield except BaseException: self.mark(stage, "raised") raise else: self.mark(stage, "returned") def close(self): if self.fd is not None: fd, self.fd = self.fd, None try: os.close(fd) except OSError: self.failure = "DIAGNOSTICS_CLOSE_FAILED"