02 Mar 2025 8 min read python

[진행중] 동적 라우팅 서버 개발 로그

개요

작업: 무중단 동적 라우팅 서버 개발
시작일: 2025.02.17 ~

개발환경

Python 3.11
FastAPI
Watchdog
Redis
Code Server

세부사항

1) 동적 라우팅 서버 개발

하나의 컨테이너에 3개의 동적라우트를 서빙

AI Agent API 등록/관리 서버

역할: 생성형 AI 챗봇의 응답 API를 사용자 정의한 요청/응답 스키마로 wrapping 한 API를 에이전트(agent_id) 별로 생성 및 호스팅한다.
포트: 9999
기능:
1. [POST] 동적 라우트 생성
  1. 사용자가 정의한 api 스펙(agent_id, 엔드포인트, 요청/응답 스키마, 접근유형, client_info 등)을 기반으로 동적 코드 생성 및 파일을 지정된 디스크 경로에 저장: {agent_id}.py
  2. 생성된 동적 코드(fastAPI 라우트)를 해당 서버(main app)의 sub app으로 등록
  3. sub app 등록은 에이전트 단위로 분리: 에이전트 당 여러개의 version 의 라우트를 생성 가능
  4. 각 에이전트별 openapi.json 분리 -> swagger 접속 주소 분리: 메인 앱의 시스템 라우트는 사용자가 접근 불가해야하며, 각 에이전트 별로 작업 및 관리를 할 수 있도록 하기 위한 조치
  5. 등록된 에이전트 api의 스펙은 yaml파일로 백업본과 함께 디스크 경로에 저장 후 -> Redis 에 데이터 적재
2. [DELETE] 동적 라우트 Soft Delete
  1. 등록된 라우트를 agent_id, version 정보를 받아 해당 라우트에 대해 삭제 flag를 spec파일에 추가
3. 생성된 동적 라우트 호스팅
  1. Redis로부터 등록된 해당 agent의 스펙정보를 조회
    1. 실패시, 복구메커니즘 발동
  2. (조건1) api 호출시 요청 헤더로부터 사용자 인증정보를 받아, wrapping한 API(athena 서버)의 요청 헤더로 by-pass
  3. (조건2) 접근 타입에 따른 접근 제한: public, private, protected

public: client_info 를 요청 헤더로부터 받아서 검증 후 api 정상 동작
private: 아무 조건없이 api 정상 동작
protected: 403 접근 제한 상태

(조건3) athena 서버로부터 온 응답을 사용자 응답 스키마로 customizing
1. wrapping API(athena 서버)로부터의 응답과 해당 동적 라우트의 응답 스키마를 LLM(Lexi-hub 서버)로 프롬프트와 함께 전달
2. LLM으로부터 온 응답이 최종 응답 스키마와 일치하는지 검증 -> 응답 반환

Python Sandbox Prod 서버

역할: 사용자가 직접 파이썬 코드로 특정 기능들이 구현된 파일을 배포하면, 무중단 상태에서 동적 라우트를 생성 및 호스팅한다. 생성된 라우트들은 에이전트 단위로 분리하여 관리한다.
포트: 5678
기능:
1. [POST] 동적 라우트 배포(변화감지: 생성/수정/삭제)
  1. 생성/업데이트된 파이썬 파일들을 실시간으로 감지하여, Pending 목록에서 대기
  2. Pending 목록의 기능들은 모두 Python Sandbox Dev 서버에서 정상적으로 작동하는지 테스트가 된 이후 복사된 파일임
  3. Pending 목록의 기능들은 에이전트별(디렉토리 단위) sub_app으로 분리되고, 해당 sub_app의 라우트로 등록 -> /agents/{에이전트명}/{파일명}/{엔드포인트 path}
2. 등록된 라우트 호스팅
  1. 모든 등록된 라우트는 각 에이전트별로 openapi.json이 분리된 상태
  2. 사용자가 작성한 기능은 Python Sandbox Dev 서버에서 모두 테스트가 완료되었으므로 정상적으로 작동해야함

Python Sandbox Dev 서버 (테스트 서버)

역할: Python Sandbox Prod 서버에 동적라우트를 등록하기 전, Code Server에서 작성된 파이썬 코드 기반의 기능이 무중단 실시간으로 라우트 등록/업데이트/삭제 등 반영이 되며, 이 라우트들을 테스트 할 수 있다. 또한, 에이전트 별 등록된 라우트 정보 리스트를 조회할 수 있다.
포트: 5555
기능:
1. 실시간 동적 라우트 생성/업데이트/삭제 (feat. watchdog)
  1. watchdog 감시를 통해 실시간으로 파이썬 파일에 작성된 api를 디렉토리별로 sub_app의 라우트로 생성/업데이트/삭제를 반영한다.
  2. 실시간 반영 특성상, 특정 파일의 특정 함수에 에러가 발생하는 경우에는 전체 파일 내 모든 라우트를 사용불가한 상태가 된다.
  3. 막 생성/업데이트된 라우트는 상태가 'unknown' 이다.
2. [POST] 실시간 동적 라우트 테스트
  1. 실시간으로 등록된 모든 라우트들을 테스트하고 결과를 by-pass 한다.
  2. 해당 라우트의 테스트가 통과하면, 라우트의 상태는 "stable"로 변경, 실패하면 "unstable"로 변경
3. [GET] 실시간 동적 라우트 정보 목록 조회
  1. 등록된 모든 동적 라우트의 스펙정보(파일명 및 라우트 상태정보 포함)를 반환한다.

Redis 서버

역할: 각 FastAPI Server들에서 생성되는 동적 라우트를 호출하기 위해 필요한 정보를 저장
포트: 6379
기능:
1. Agent Server:
  1. 동적 라우트 스펙정보 저장 및 전달
  2. ./routes/agents/apis/ 아래 {agent_id}_spec.yaml, {agent_id}_spec.yaml.bak 을 읽

code server 연동

실시간 반영 코드 작성 기능
파이썬 코드 스니펫 기능
코드 테스트 & 디버깅 기능

2) 로거 개발

이 태스크는 py-runner 애플리케이션의 로깅 시스템을 APM(Application Performance Monitoring) 도구와 통합하기 위함

태스크 요약

애플리케이션 로그를 포지큐브 표준 포맷의 JSON으로 변환하여 효과적인 모니터링과 디버깅을 가능하게 하는 작업

요구사항 분석

로그 형식 변환: 현재 py-runner의 컨테이너 로그를 포지큐브 표준 입출력 형태(JSON)로 변환
필수 포함 정보:
- API 요청(request) 정보: header, path, origin, x-forwarded-for, remoteAddress 등
- API 응답(response) 정보: HTTP 상태 코드, 지연 시간(latency), 메시지

구현 계획

1. 로깅 미들웨어 구현

py-runner는 FastAPI를 사용하고 있으므로, FastAPI의 미들웨어 기능을 활용

기존 코드를 보면 이미 ExceptionMiddleware가 구현되어 있으므로, 비슷한 방식으로 로깅 미들웨어를 추가하면 될듯

# logging_middleware.py (새 파일)
import time
import json
import logging
import socket
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp

logger = logging.getLogger("[API-LOGGER]")

class APILoggingMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp):
        super().__init__(app)
        self.hostname = socket.gethostname()
        
    async def dispatch(self, request: Request, call_next):
        # 요청 시작 시간 기록
        start_time = time.time()
        request_id = request.headers.get("x-request-id", "")
        
        # 응답 처리
        response = await call_next(request)
        
        # 응답 시간 계산 (밀리초)
        response_time = int((time.time() - start_time) * 1000)
        
        # 로그 데이터 구성
        log_data = {
            "level": "INFO",
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S.%fZ", time.gmtime()),
            "pid": os.getpid(),
            "hostname": self.hostname,
            "req": {
                "id": id(request),
                "method": request.method,
                "url": str(request.url),
                "query": dict(request.query_params),
                "headers": dict(request.headers),
                "remoteAddress": request.client.host if request.client else "",
                "remotePort": request.client.port if request.client else 0
            },
            "res": {
                "statusCode": response.status_code,
                "headers": dict(response.headers)
            },
            "responseTime": response_time,
            "module": "py-runner",
            "message": "request completed"
        }
        
        # JSON으로 직렬화하여 로그 출력
        logger.info(json.dumps(log_data))
        
        return response

2. 미들웨어 등록

세 가지 애플리케이션(agent, sandbox, dev)에 미들웨어를 각각 등록

# apps/agent.py, apps/sandbox.py, apps/dev.py 각각에 추가
from utils.logging_middleware import APILoggingMiddleware

# 다른 미들웨어보다 먼저 추가하는 것이 좋음
app.add_middleware(APILoggingMiddleware)

3. 로깅 설정 강화

로깅 설정을 조정하여 로그 포맷을 지정하고 출력 방식을 설정:

# 로깅 설정 함수 (utils/__init__.py나 새로운 모듈에 추가)
def configure_api_logging():
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "json": {
                "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
                "fmt": "%(levelname)s %(asctime)s %(message)s"
            }
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "formatter": "json"
            }
        },
        "loggers": {
            "[API-LOGGER]": {
                "handlers": ["console"],
                "level": "INFO"
            }
        }
    })

구현 단계

필요한 종속성 추가: python-json-logger 패키지를 pyproject.toml에 이미 있는지 확인 (있음)
로깅 미들웨어 구현: 위의 코드를 참조하여 구현
각 애플리케이션에 미들웨어 등록: 세 가지 앱 모두에 미들웨어 추가
로깅 설정 통합: main.py에서 서버 시작 전에 로깅 설정을 적용
테스트 및 검증: 각 서버를 실행하고 API 요청을 보내 로그 포맷이 요구사항과 일치하는지 확인

추가 코드

다음은 로깅 미들웨어 구현 추가 코드:

# utils/headers_util.py (새 파일)
def extract_trace_info(headers):
    """
    트레이싱 관련 헤더 정보 추출
    """
    trace_id = ""
    span_id = ""
    trace_flags = "00"
    
    # B3 포맷
    if "x-b3-traceid" in headers:
        trace_id = headers.get("x-b3-traceid")
        span_id = headers.get("x-b3-spanid", "")
        trace_flags = "01" if headers.get("x-b3-sampled") == "1" else "00"
    
    # W3C 트레이스컨텍스트 포맷
    elif "traceparent" in headers:
        traceparent = headers.get("traceparent", "")
        parts = traceparent.split("-")
        if len(parts) >= 3:
            trace_id = parts[1]
            span_id = parts[2]
            trace_flags = parts[0]
    
    return {
        "trace_id": trace_id,
        "span_id": span_id, 
        "trace_flags": trace_flags
    }

개요