diff --git a/ai/src/ai_server/CLAUDE.md b/ai/src/ai_server/CLAUDE.md index 6a554d58..65910802 100644 --- a/ai/src/ai_server/CLAUDE.md +++ b/ai/src/ai_server/CLAUDE.md @@ -95,6 +95,14 @@ async def consume(message: AbstractIncomingMessage) -> None: await mark_processed(envelope.message_id) ``` +> **`trace_context` 는 2026-09-28 까지 이 문서에만 있었고 구현이 없었다.** 이 패턴을 따라 +> 쓰면 ImportError 가 났고, 실제 컨슈머들은 traceId 를 로그에 전혀 싣지 않았다 — 운영 AI +> 로그에 trace 가 하나도 없던 이유다. 지금은 `observability/trace.py` 에 있고 +> `consume_with_failure_signal`(생성·분석 컨슈머 공통) + `voice`/`tts` 컨슈머가 쓴다. +> structlog 기본 프로세서 첫 번째가 `merge_contextvars` 라 한 번 묶으면 그 구간의 **모든 +> 로그 줄에 자동으로 실린다**. contextvars 는 asyncio 태스크별로 격리돼 동시 처리 메시지끼리 +> 섞이지 않는다(`tests/test_trace_context.py` 가 고정). + ### `analyzer/` - 분석 use case 단위 (`resume_analyzer.py`, `repository_analyzer.py`, `web_resume_analyzer.py`) - 소스 추출 추상화는 `analyzer/sources/` (PDF/GitHub/웹/텍스트), 임베딩 인제스트는 `_embedding_step.py` diff --git a/ai/src/ai_server/messaging/consumers/failure_signal.py b/ai/src/ai_server/messaging/consumers/failure_signal.py index acf0fe3c..e448081d 100644 --- a/ai/src/ai_server/messaging/consumers/failure_signal.py +++ b/ai/src/ai_server/messaging/consumers/failure_signal.py @@ -8,6 +8,7 @@ from aio_pika.abc import AbstractIncomingMessage from pydantic import BaseModel +from ai_server.observability.trace import trace_context from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope @@ -118,68 +119,69 @@ async def consume_with_failure_signal( ) raise - if idempotency.is_seen_then_mark(envelope.message_id): - log.info( - f"{domain}.idempotent.skip", - message_id=envelope.message_id, - trace_id=envelope.trace_id, - ) - return - - async def _publish(payload: BaseModel) -> None: - await publisher.publish( - routing_key=routing_key, - message_type=message_type, - payload=payload, - trace_id=envelope.trace_id, - correlation_id=envelope.message_id, - context=envelope.context, - ) + with trace_context(envelope.trace_id): + if idempotency.is_seen_then_mark(envelope.message_id): + log.info( + f"{domain}.idempotent.skip", + message_id=envelope.message_id, + trace_id=envelope.trace_id, + ) + return + + async def _publish(payload: BaseModel) -> None: + await publisher.publish( + routing_key=routing_key, + message_type=message_type, + payload=payload, + trace_id=envelope.trace_id, + correlation_id=envelope.message_id, + context=envelope.context, + ) - log_ids: dict[str, Any] = { - "message_id": envelope.message_id, - "trace_id": envelope.trace_id, - } - session_id = getattr(envelope.payload, "session_id", None) - if session_id is not None: - log_ids["session_id"] = session_id + log_ids: dict[str, Any] = { + "message_id": envelope.message_id, + "trace_id": envelope.trace_id, + } + session_id = getattr(envelope.payload, "session_id", None) + if session_id is not None: + log_ids["session_id"] = session_id - try: - payload = await process(envelope) - except Exception as exc: # noqa: BLE001 - if expected_errors and isinstance(exc, expected_errors): - # 예상된 도메인 실패(빈 PDF·404 URL 등 일상 입력)는 traceback 없는 warning — - # ERROR 레벨 스택트레이스로 일상 실패가 알람을 울리지 않게 한다. - log.warning( - f"{domain}.{action}.failed", - error_code=getattr(exc, "code", None), - retriable=getattr(exc, "retriable", None), - error=str(exc), - **log_ids, - ) - else: - log.exception(f"{domain}.{action}.failed", **log_ids) try: - # 팩토리 자체가 죽어도(검증 오류 등) 같은 안전망을 태운다 — - # try 밖이면 unmark 없이 DLQ 로 가서 재주입이 duplicate skip 으로 삼켜진다. - fallback = failed_payload(envelope.payload, exc) - await _publish(fallback) - except Exception: # noqa: BLE001 - log.exception( - f"{domain}.failed_callback.publish_failed", - **log_ids, - ) - idempotency.unmark(envelope.message_id) - raise exc - return + payload = await process(envelope) + except Exception as exc: # noqa: BLE001 + if expected_errors and isinstance(exc, expected_errors): + # 예상된 도메인 실패(빈 PDF·404 URL 등 일상 입력)는 traceback 없는 warning — + # ERROR 레벨 스택트레이스로 일상 실패가 알람을 울리지 않게 한다. + log.warning( + f"{domain}.{action}.failed", + error_code=getattr(exc, "code", None), + retriable=getattr(exc, "retriable", None), + error=str(exc), + **log_ids, + ) + else: + log.exception(f"{domain}.{action}.failed", **log_ids) + try: + # 팩토리 자체가 죽어도(검증 오류 등) 같은 안전망을 태운다 — + # try 밖이면 unmark 없이 DLQ 로 가서 재주입이 duplicate skip 으로 삼켜진다. + fallback = failed_payload(envelope.payload, exc) + await _publish(fallback) + except Exception: # noqa: BLE001 + log.exception( + f"{domain}.failed_callback.publish_failed", + **log_ids, + ) + idempotency.unmark(envelope.message_id) + raise exc + return - try: - await _publish(payload) - except Exception: - idempotency.unmark(envelope.message_id) - raise - log.info( - f"{domain}.{action}.done", - **log_ids, - **(done_fields(payload) if done_fields is not None else {}), - ) + try: + await _publish(payload) + except Exception: + idempotency.unmark(envelope.message_id) + raise + log.info( + f"{domain}.{action}.done", + **log_ids, + **(done_fields(payload) if done_fields is not None else {}), + ) diff --git a/ai/src/ai_server/messaging/consumers/tts_consumer.py b/ai/src/ai_server/messaging/consumers/tts_consumer.py index a56ca617..16aae488 100644 --- a/ai/src/ai_server/messaging/consumers/tts_consumer.py +++ b/ai/src/ai_server/messaging/consumers/tts_consumer.py @@ -6,6 +6,7 @@ import structlog from aio_pika.abc import AbstractIncomingMessage +from ai_server.observability.trace import trace_context from ai_server.chain.sentence_split import next_sentences from ai_server.messaging.consumers.failure_signal import unmark_on_error from ai_server.messaging.idempotency import LruIdempotencyStore @@ -81,65 +82,66 @@ async def handle(self, message: AbstractIncomingMessage) -> None: ) raise - if self._idempotency.is_seen_then_mark(envelope.message_id): - log.info("tts.idempotent.skip", message_id=envelope.message_id) - return + with trace_context(envelope.trace_id): + if self._idempotency.is_seen_then_mark(envelope.message_id): + log.info("tts.idempotent.skip", message_id=envelope.message_id) + return - # 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6). - async with unmark_on_error(self._idempotency, envelope.message_id): - req = envelope.payload - log.info( - "tts.synthesize.start", - message_id=envelope.message_id, - session_id=req.session_id, - interview_message_id=req.message_id, - streaming=self._notifier is not None, - trace_id=envelope.trace_id, - ) + # 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6). + async with unmark_on_error(self._idempotency, envelope.message_id): + req = envelope.payload + log.info( + "tts.synthesize.start", + message_id=envelope.message_id, + session_id=req.session_id, + interview_message_id=req.message_id, + streaming=self._notifier is not None, + trace_id=envelope.trace_id, + ) - if self._notifier is not None: - result = await self._synthesize_segmented(envelope, req) - else: - result = await self._synthesize_once(envelope, req) - if result is None: - return # 실패 — 콜백은 내부에서 이미 발행 + if self._notifier is not None: + result = await self._synthesize_segmented(envelope, req) + else: + result = await self._synthesize_once(envelope, req) + if result is None: + return # 실패 — 콜백은 내부에서 이미 발행 - key = self._build_key( - req.session_id, req.message_id, result.content_type - ) - try: - await self._storage.put_bytes( - key, result.audio_bytes, content_type=result.content_type + key = self._build_key( + req.session_id, req.message_id, result.content_type ) - except Exception as exc: - log.error("tts.storage.failed", error=str(exc), key=key) + try: + await self._storage.put_bytes( + key, result.audio_bytes, content_type=result.content_type + ) + except Exception as exc: + log.error("tts.storage.failed", error=str(exc), key=key) + await self._publish( + envelope, + TtsCallbackPayload( + session_id=req.session_id, + message_id=req.message_id, + status="FAILED", + error_code="TTS_STORAGE_FAILED", + ), + ) + return + await self._publish( envelope, TtsCallbackPayload( session_id=req.session_id, message_id=req.message_id, - status="FAILED", - error_code="TTS_STORAGE_FAILED", + status="SUCCEEDED", + audio_key=key, + duration_sec=result.duration_sec, ), ) - return - - await self._publish( - envelope, - TtsCallbackPayload( + log.info( + "tts.synthesize.done", session_id=req.session_id, - message_id=req.message_id, - status="SUCCEEDED", - audio_key=key, - duration_sec=result.duration_sec, - ), - ) - log.info( - "tts.synthesize.done", - session_id=req.session_id, - interview_message_id=req.message_id, - key=key, - ) + interview_message_id=req.message_id, + key=key, + ) async def _synthesize_once( self, envelope: Envelope[GenerateTtsRequest], req: GenerateTtsRequest diff --git a/ai/src/ai_server/messaging/consumers/voice_consumer.py b/ai/src/ai_server/messaging/consumers/voice_consumer.py index ef462c34..15225537 100644 --- a/ai/src/ai_server/messaging/consumers/voice_consumer.py +++ b/ai/src/ai_server/messaging/consumers/voice_consumer.py @@ -6,6 +6,7 @@ import structlog from aio_pika.abc import AbstractIncomingMessage +from ai_server.observability.trace import trace_context from ai_server.core.client import CoreClient from ai_server.messaging.consumers.failure_signal import unmark_on_error from ai_server.messaging.idempotency import LruIdempotencyStore @@ -71,77 +72,80 @@ async def handle(self, message: AbstractIncomingMessage) -> None: ) raise - if self._idempotency.is_seen_then_mark(envelope.message_id): - log.info("voice.idempotent.skip", message_id=envelope.message_id) - return - - # 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6). - async with unmark_on_error(self._idempotency, envelope.message_id): - req = envelope.payload - log.info( - "voice.analyze.start", - message_id=envelope.message_id, - session_id=req.session_id, - interview_message_id=req.message_id, - key=req.audio_s3_key, - trace_id=envelope.trace_id, - ) - - try: - audio_bytes = await self._storage.get_bytes(req.audio_s3_key) - except Exception as exc: - log.error( - "voice.storage.failed", error=str(exc), key=req.audio_s3_key - ) - await self._publish_failed(envelope, req, code="AUDIO_FETCH_FAILED") + with trace_context(envelope.trace_id): + if self._idempotency.is_seen_then_mark(envelope.message_id): + log.info("voice.idempotent.skip", message_id=envelope.message_id) return - try: - result = await self._transcribe_with_retry( - req, envelope, audio_bytes - ) - except SttError as exc: - log.error( - "voice.stt.failed", - error=str(exc), - code=exc.code, + # 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6). + async with unmark_on_error(self._idempotency, envelope.message_id): + req = envelope.payload + log.info( + "voice.analyze.start", + message_id=envelope.message_id, session_id=req.session_id, - attempts=getattr(exc, "attempts", 1), + interview_message_id=req.message_id, + key=req.audio_s3_key, + trace_id=envelope.trace_id, ) - await self._publish_failed(envelope, req, code=exc.code) - return - except Exception as exc: - log.error( - "voice.stt.unexpected", - error=str(exc), + + try: + audio_bytes = await self._storage.get_bytes(req.audio_s3_key) + except Exception as exc: + log.error( + "voice.storage.failed", error=str(exc), key=req.audio_s3_key + ) + await self._publish_failed( + envelope, req, code="AUDIO_FETCH_FAILED" + ) + return + + try: + result = await self._transcribe_with_retry( + req, envelope, audio_bytes + ) + except SttError as exc: + log.error( + "voice.stt.failed", + error=str(exc), + code=exc.code, + session_id=req.session_id, + attempts=getattr(exc, "attempts", 1), + ) + await self._publish_failed(envelope, req, code=exc.code) + return + except Exception as exc: + log.error( + "voice.stt.unexpected", + error=str(exc), + session_id=req.session_id, + ) + await self._publish_failed( + envelope, req, code="TRANSCRIPTION_FAILED" + ) + return + + metrics = analyze(result, filler_pattern=self._filler_pattern) + + payload = VoiceCallbackPayload( session_id=req.session_id, + interview_message_id=req.message_id, + transcript=result.text, + speaking_rate_wpm=metrics.speaking_rate_wpm, + silence_duration_sec=metrics.silence_duration_sec, + filler_word_counts=metrics.filler_word_counts, + pronunciation_accuracy=metrics.pronunciation_accuracy, + error_code=None, ) - await self._publish_failed( - envelope, req, code="TRANSCRIPTION_FAILED" + await self._publish_callback(envelope, payload) + log.info( + "voice.analyze.done", + message_id=envelope.message_id, + session_id=req.session_id, + interview_message_id=req.message_id, + wpm=metrics.speaking_rate_wpm, + trace_id=envelope.trace_id, ) - return - - metrics = analyze(result, filler_pattern=self._filler_pattern) - - payload = VoiceCallbackPayload( - session_id=req.session_id, - interview_message_id=req.message_id, - transcript=result.text, - speaking_rate_wpm=metrics.speaking_rate_wpm, - silence_duration_sec=metrics.silence_duration_sec, - filler_word_counts=metrics.filler_word_counts, - pronunciation_accuracy=metrics.pronunciation_accuracy, - error_code=None, - ) - await self._publish_callback(envelope, payload) - log.info( - "voice.analyze.done", - message_id=envelope.message_id, - session_id=req.session_id, - interview_message_id=req.message_id, - wpm=metrics.speaking_rate_wpm, - trace_id=envelope.trace_id, - ) async def _transcribe_with_retry( self, req: AnalyzeVoiceRequest, envelope, audio_bytes: bytes diff --git a/ai/src/ai_server/observability/trace.py b/ai/src/ai_server/observability/trace.py new file mode 100644 index 00000000..c4d61d13 --- /dev/null +++ b/ai/src/ai_server/observability/trace.py @@ -0,0 +1,39 @@ +"""메시지 처리 구간에 traceId 를 묶는다. + +`ai/CLAUDE.md` 는 오래 전부터 컨슈머 패턴으로 이렇게 적어 왔다:: + + with trace_context(envelope.trace_id): + await resume_analyzer.handle(envelope.payload) + +그런데 **`trace_context` 가 존재한 적이 없다.** 문서를 따라 쓰면 ImportError 가 나고, +실제 컨슈머들은 traceId 를 로그에 전혀 싣지 않았다 — 운영 AI 로그에 trace 가 하나도 +없는 이유다. Core 는 2026-09-28 에 같은 문제를 고쳤고(로그 패턴·필터 순서·큐 전파), +이건 그 나머지 절반이다. + +structlog 기본 프로세서 체인의 첫 번째가 ``merge_contextvars`` 라서, 여기서 한 번 묶어 +두면 **그 구간의 모든 로그 줄에 자동으로 실린다** — 호출부마다 인자로 넘길 필요가 없다. +contextvars 는 asyncio 태스크별로 격리되므로 동시에 처리되는 메시지끼리 섞이지 않는다. +""" + +from __future__ import annotations + +import contextlib +import uuid +from collections.abc import Iterator + +import structlog + + +@contextlib.contextmanager +def trace_context(trace_id: str | None) -> Iterator[str]: + """`trace_id` 를 이 구간의 모든 로그에 바인딩한다. 실제 쓰인 값을 돌려준다. + + 비어 있으면 새로 만든다 — 없는 것보다 낫고, 적어도 그 처리 한 건은 묶인다. + 끝나면 반드시 되돌린다: 워커 태스크가 재사용되면 남은 값이 다음 메시지 로그에 묻는다. + """ + effective = trace_id if trace_id else str(uuid.uuid4()) + tokens = structlog.contextvars.bind_contextvars(trace_id=effective) + try: + yield effective + finally: + structlog.contextvars.reset_contextvars(**tokens) diff --git a/ai/tests/test_trace_context.py b/ai/tests/test_trace_context.py new file mode 100644 index 00000000..6d9a273e --- /dev/null +++ b/ai/tests/test_trace_context.py @@ -0,0 +1,88 @@ +"""traceId 가 **실제로 로그 줄에 실리는지** 확인한다. + +Core 쪽에서 똑같은 함정을 겪었다: MDC 는 채우는데 로그 패턴이 참조하지 않아 +운영 로그 어디에도 traceId 가 없었다. "바인딩했다"가 아니라 "출력에 나온다"를 본다. +""" + +from __future__ import annotations + +import asyncio + +import structlog + +from ai_server.observability.trace import trace_context + + +def _capture(): + cap = structlog.testing.LogCapture() + structlog.configure( + processors=[structlog.contextvars.merge_contextvars, cap], + ) + return cap + + +def teardown_function() -> None: + structlog.reset_defaults() + structlog.contextvars.clear_contextvars() + + +def test_trace_id_appears_in_log_output(): + cap = _capture() + log = structlog.get_logger("t") + + with trace_context("envelope-trace-1"): + log.info("inside") + log.info("outside") + + assert cap.entries[0]["trace_id"] == "envelope-trace-1" + # 빠져나오면 지워져야 한다 — 남으면 다음 메시지 로그에 엉뚱한 traceId 가 묻는다. + assert "trace_id" not in cap.entries[1] + + +def test_trace_id_generated_when_missing(): + cap = _capture() + log = structlog.get_logger("t") + + with trace_context(None) as a: + log.info("a") + with trace_context("") as b: + log.info("b") + + # 없는 것보다 낫다 — 적어도 그 처리 한 건은 묶인다. + assert a and b and a != b + assert cap.entries[0]["trace_id"] == a + assert cap.entries[1]["trace_id"] == b + + +def test_trace_id_cleared_on_exception(): + cap = _capture() + log = structlog.get_logger("t") + + try: + with trace_context("t1"): + raise RuntimeError("boom") + except RuntimeError: + pass # 컨슈머는 예외를 다시 던져 DLQ 로 보낸다 — 그 경로에서도 정리돼야 한다 + log.info("after") + + assert "trace_id" not in cap.entries[0] + + +def test_concurrent_messages_do_not_leak_into_each_other(): + """컨슈머는 메시지를 동시에 처리한다. contextvars 가 태스크별로 격리되지 않으면 + A 의 traceId 가 B 의 로그에 묻는다 — 추적이 틀린 것은 없는 것보다 나쁘다.""" + cap = _capture() + log = structlog.get_logger("t") + + async def work(tid: str, delay: float) -> None: + with trace_context(tid): + await asyncio.sleep(delay) + log.info("done", which=tid) + + async def main() -> None: + await asyncio.gather(work("A", 0.02), work("B", 0.01), work("C", 0.0)) + + asyncio.run(main()) + + seen = {e["which"]: e["trace_id"] for e in cap.entries} + assert seen == {"A": "A", "B": "B", "C": "C"} diff --git a/docs/observability.md b/docs/observability.md index 7e850e85..4e8a9827 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -13,7 +13,10 @@ 되돌리지 않아, 이어서 발행하는 메시지에 `RabbitMessagePublisher` 가 **새 traceId** 를 붙였다. 지금은 5개 `@RabbitListener` 가 `TraceContext.runWithTraceId` 로 감싸 원 요청의 id 아래에서 실행된다. - - AI 서버 로그에는 아직 traceId 가 없다(uvicorn 기본 액세스 로그). 남은 과제. + - AI 서버도 2026-09-28 부터 싣는다 — `observability/trace.py` 의 `trace_context` 가 + 엔벨로프 traceId 를 structlog contextvars 에 묶어 그 구간 모든 로그에 자동 부착한다. + (그전까지는 `ai/CLAUDE.md` 가 `trace_context` 를 패턴으로 제시했는데 **구현이 없었다**.) + uvicorn 의 액세스 로그(`GET /health`)에는 여전히 없다 — 요청 단위 미들웨어는 미도입. ### 1.2 전파 규약