[Websocket] 웹소켓 로그 ai search에 적재: 스트리밍 단일 관리

🧩 배경 설명

1. 일단 내가 지금 몸 담고 있는 플젝이 REST 로그 적재는 구현되어 있는데 ws는 안 되어 있음

2. 그래서 해야 됨

3. 시도

4. 그런데 새로 붙인 웹소켓 서비스는 로그 수집 구조가 전혀 다름을 깨달음(당연함)

 

 

 

💭 고민의 내용

send_json() 이 호출될 때마다 로그가 수백 건씩 쌓임.

resToken이 738인 응답이 스트리밍 되면 ,,, 적어도 738+@ 건이 쌓이는 거임 그 하나의 리시브마다;;

REST처럼 최종 하나만 남겨야 됨

 

 

🛠️ 해결을 위한 시도들

1. 구글링

검색 키워드: FastAPI Websocket, Logging, Websocket 로그 적재

간지러운 부분을 딱 긁어줄 기술블로그를 찾을 수 없었음 

 

2. 직접 시도

• Tabp 클래스 만들어서 send_json() 을 가로챔
• 마지막 페이로드만 저장해서 단일 적재 구현함
• _logOnly 플래그로 클라 전송을 차단

# ─────────────────────────────────────────────
# 0) Tap: 최종 페이로드만 보존 (statusCode/statusMsg/result)
# ─────────────────────────────────────────────
class WebSocketTap:
    def __init__(self, websocket):
        self._ws = websocket
        self.final_payload = None  # {"statusCode":..,"statusMsg":..,"result":..}

    async def send_json(self, data, *args, **kwargs):
        # ws_rag 내부: if websocket: await websocket.send_json({...})
        if isinstance(data, dict):
            # 매 호출마다 덮어써서 "가장 마지막" 페이로드만 유지
            self.final_payload = {
                "statusCode": data.get("statusCode"),
                "statusMsg":  data.get("statusMsg"),
                "result":     data.get("result"),
            }
        return await self._ws.send_json(data, *args, **kwargs)

    # 필요하면 text도 오버라이드 가능하지만, 여기선 최종 json만 대상
    def __getattr__(self, name):
        return getattr(self._ws, name)

# ─────────────────────────────────────────────
# 1) 로그 편집 → 최종 1건만 insert 호출 → ai search에 적재
# ─────────────────────────────────────────────
async def websocketCallLog(logData: dict):
    doc = {
        "event":        logData.get("event"),
        "userId":       logData.get("userId"),
        "requestKey":   logData.get("requestKey"),
        "infId":        logData.get("infId"),
        "inData":       logData.get("inData"),
        "statusCode":   logData.get("statusCode"),
        "statusMsg":    logData.get("statusMsg"),
        "tapResult":    logData.get("tapResult"),     # if websocket: send_json(...)의 result
        "methodResult": logData.get("methodResult"),  # getStrm() 리턴
    }
    await insertWebsocketCallLog(doc)  # ai search에 적재


# ─────────────────────────────────────────────
# 2) 서비스 태스크: Tap으로 감싸 호출 → 마지막 페이로드만 로그 1건
# ─────────────────────────────────────────────
async def websocketSendTextAsync(websocket, data):
    ws_tap = WebSocketTap(websocket)
    method_result = None
    try:
        # 특정 인터페이스 작업 후 result
        method_result = await ws_rag.getStrm(event, data, ws_tap)
        
    finally:
        # 최종 페이로드가 없을 수도 있으니 방어 로직도 필요할 듯함
        final = ws_tap.final_payload or {}
        
        # ai search에 넣을 데이터 편집해서 던져주기!
        logData = {
            "event":        event,
            "userId":       f"{websocket.client.host}:{websocket.client.port}",
            "requestKey":   req_key,
            "infId":        inf_id,
            "inData":       data,
            "statusCode":   final.get("statusCode"),
            "statusMsg":    final.get("statusMsg"),
            "tapResult":    final.get("result"),
            "methodResult": method_result,
        }
        # 반드시 한 번만 적재
        await websocketCallLog(logData)


# ─────────────────────────────────────────────
# 3) 엔드포인트: 멀티면 태스크, 단건이면 await (마지막에 close)
# ─────────────────────────────────────────────
@app.websocket("/ws/rag")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            raw = await websocket.receive_text()
            if raw == "__ping__":
                await websocket.send_text("pong")
                continue

            try:
                data = json.loads(raw)
                assert isinstance(data, dict)
            except Exception as e:
                await websocket.send_json({"statusCode": 400, "statusMsg": f"INVALID_JSON: {e}"})
                continue

            multi = data.get("multi")
            if multi:
                # 멀티여도 각 요청별로 최종 1건만 적재됨
                asyncio.create_task(websocketSendTextAsync(websocket, data))
            else:
                await websocketSendTextAsync(websocket, data)

    finally:
        try:
            await websocket.close()
        except Exception:
            pass

대충 이런 로직이였음

 

📈 변화

• 내가 원하는 대로 됨
  • 로그 적재량도 많이 줄이고(약 90% 이상 감소)
  • 화면에는 정상 응답만 보이게!
  • 디버그/에러는 인덱스에서만 볼 수 있음
• 이 플젝이 갖고 있는 REST 로그 적재 구조와 어느 정도 일관성을 가지게끔 함(메서드명, 플로우 등)

 

📌 회고 & 다음 목표

웹소켓이 REST와 본질적으로 다른 구조라 "REST처럼 하겠다"는 발상이 문제였던 것 같음

이번 포스팅에서 REST 로직 구조를 설명하진 않았지만 REST 로직을 참고하여 ws에 맞게 리디자인 한 것이 재밌었음

서비스 성격에 맞춰 커스텀해야 한다는 것이 이번 과업을 통해 얻은 교훈임!