panopticon-monitoring 0.1.3

FastAPI monitoring SDK for logs and distributed tracing

@panopticon/fastapi-monitoring-sdk

간편한 FastAPI 모니터링 SDK - 로그와 분산 추적을 자동 수집

특징

• 간편한 통합: 간단한 설정만으로 완벽한 모니터링 시스템 구축
• 자동 수집: HTTP 요청, 로그, 외부 API 호출, Bedrock API 호출 자동 추적
• 분산 추적: OpenTelemetry 호환 Trace ID/Span ID 자동 생성
• 최소 오버헤드: 배치 전송으로 애플리케이션 성능 영향 최소화
• Non-blocking: 모니터링 실패가 앱 동작에 영향 없음
• Type Hints: 완전한 타입 힌트 지원

설치

pip install panopticon-monitoring

빠른 시작

기본 설정

# main.py
from fastapi import FastAPI
from panopticon_monitoring import MonitoringSDK

app = FastAPI()

# 이 코드만 추가하세요!
MonitoringSDK.init(app, {
    'api_key': 'your-api-key-123',
    'endpoint': 'https://producer.woongno-monitoring.com',
    'service_name': 'llm-service',
    'environment': 'production'
})

@app.get("/")
async def root():
    return {"message": "Hello World"}

이게 전부입니다! 이제 다음이 자동으로 수집됩니다:

• ✅ 모든 HTTP 요청/응답 (Root Span)
• ✅ 애플리케이션 로그 (Python logging)
• ✅ 외부 API 호출 (httpx, requests)
• ✅ Bedrock API 호출 (boto3)

수집 데이터

1. HTTP 요청 Trace (Root Span)

{
  "type": "span",
  "timestamp": "2025-11-23T10:03:10.947563Z",
  "service_name": "llm-service",
  "environment": "production",
  "trace_id": "ceed8e5abfdfec0d6cc06b9eb8e53005",
  "span_id": "9678cbd561a00b1d",
  "parent_span_id": null,
  "name": "POST /chat/completions",
  "kind": "SERVER",
  "duration_ms": 1250.5,
  "status": "OK",
  "http_method": "POST",
  "http_path": "/chat/completions",
  "http_status_code": 200
}

2. 외부 API 호출 Trace (Child Span)

{
  "type": "span",
  "trace_id": "ceed8e5abfdfec0d6cc06b9eb8e53005",
  "span_id": "abc123def456",
  "parent_span_id": "9678cbd561a00b1d",
  "name": "POST https://api.nestjs-service.com/users",
  "kind": "CLIENT",
  "duration_ms": 45.2,
  "http_method": "POST",
  "http_url": "https://api.nestjs-service.com/users",
  "http_status_code": 200
}

3. Bedrock API 호출 Trace (Child Span)

{
  "type": "span",
  "trace_id": "ceed8e5abfdfec0d6cc06b9eb8e53005",
  "span_id": "def789ghi012",
  "parent_span_id": "9678cbd561a00b1d",
  "name": "Bedrock InvokeModel",
  "kind": "CLIENT",
  "duration_ms": 1180.3,
  "bedrock_model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
  "bedrock_operation": "InvokeModel",
  "bedrock_input_tokens": 150,
  "bedrock_output_tokens": 420
}

4. 로그

{
  "type": "log",
  "timestamp": "2025-11-23T10:03:10.947563Z",
  "service_name": "llm-service",
  "environment": "production",
  "level": "info",
  "message": "Processing chat completion request",
  "context": "app.routes.chat"
}

설정 옵션

from typing import TypedDict, Optional

class MonitoringConfig(TypedDict):
    # 필수 설정
    api_key: str              # Producer 서버 인증 키
    service_name: str         # 서비스 이름

    # 엔드포인트 설정 (둘 중 하나 필수)
    endpoint: Optional[str]   # Producer 서버 URL (자동으로 /sdk/logs, /sdk/traces 추가)
    # 또는 개별 지정
    log_endpoint: Optional[str]    # 로그 전송 URL
    trace_endpoint: Optional[str]  # 트레이스 전송 URL

    # 선택 설정
    environment: Optional[str]        # 환경 (기본: 'development')
    batch_size: Optional[int]         # 배치 크기 (기본: 100)
    flush_interval: Optional[int]     # 전송 주기 초 (기본: 5)

    # 기능 토글
    enable_log_tracking: Optional[bool]          # 로그 수집 (기본: True)
    enable_http_tracking: Optional[bool]         # HTTP 추적 (기본: True)
    enable_http_client_tracking: Optional[bool]  # 외부 API 추적 (기본: True)
    enable_bedrock_tracking: Optional[bool]      # Bedrock 추적 (기본: True)

사용 예제

FastAPI 애플리케이션과 통합

from fastapi import FastAPI
from panopticon_monitoring import MonitoringSDK
import httpx
import boto3
import logging

app = FastAPI()

# SDK 초기화
sdk = MonitoringSDK.init(app, {
    'api_key': 'your-api-key',
    'endpoint': 'https://producer.woongno-monitoring.com',
    'service_name': 'llm-service',
    'environment': 'production',
    'batch_size': 50,
    'flush_interval': 3
})

# 로거 설정
logger = logging.getLogger(__name__)

# Bedrock 클라이언트
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

@app.post("/chat/completions")
async def chat_completions(request: dict):
    logger.info("Processing chat completion request")

    # 외부 API 호출 (자동 추적됨)
    async with httpx.AsyncClient() as client:
        user_response = await client.post(
            "https://api.nestjs-service.com/users/validate",
            json={"user_id": request.get("user_id")}
        )

    # Bedrock 호출 (자동 추적됨)
    bedrock_response = bedrock.invoke_model(
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": request.get("message")}]
        })
    )

    logger.info("Chat completion successful")
    return {"response": "..."}

수동으로 Trace Context 접근

from panopticon_monitoring import get_current_trace_id, get_current_span_id

@app.get("/health")
async def health_check():
    trace_id = get_current_trace_id()
    span_id = get_current_span_id()

    logger.info(f"Health check called [TraceID: {trace_id}]")

    return {"status": "healthy", "trace_id": trace_id}

수동으로 데이터 Flush

# 테스트나 특수한 경우 즉시 전송
await sdk.flush()

# 현재 버퍼 크기 확인
buffer_size = sdk.get_buffer_size()
print(f'Buffer size: {buffer_size}')

아키텍처

FastAPI App (SDK)
    ↓
BatchSender (배치 수집)
    ↓
Producer Server (HTTP/HTTPS)
    ↓
MSK (Kafka)
    ↓
Consumer → OpenSearch/TimescaleDB

성능 최적화

• 배치 전송: 100개 또는 5초마다 일괄 전송 (설정 가능)
• Non-blocking: 전송 실패 시 앱 동작에 영향 없음
• ContextVar: Trace Context를 효율적으로 관리
• 최소 파싱: SDK는 수집만, 복잡한 파싱은 Producer에서

시스템 요구사항

• Python >= 3.8
• FastAPI >= 0.68.0
• httpx (선택사항, HTTP 클라이언트 추적용)
• requests (선택사항, HTTP 클라이언트 추적용)
• boto3 (선택사항, Bedrock 추적용)

의존성

pip install fastapi httpx boto3

라이센스

MIT

지원

이슈나 질문은 GitHub Issues에서 관리합니다.