log-collector 1.0.0

비동기 로그 수집 클라이언트

Log Collector - Python Client

고성능 비동기 로그 수집 클라이언트 for Python

✨ 주요 기능

• ⚡ 비블로킹 로깅 - 앱 블로킹 < 0.1ms
• 🚀 배치 전송 - 1000건 or 1초마다 자동 전송
• 📦 자동 압축 - gzip 압축으로 네트워크 비용 절감
• 🔄 Graceful Shutdown - 앱 종료 시 큐 자동 flush
• 🎯 자동 필드 수집 - 호출 위치, HTTP 컨텍스트, 사용자 컨텍스트 자동 포함
• 🌐 웹 프레임워크 통합 - Flask, FastAPI, Django 지원
• 🔍 분산 추적 - trace_id로 마이크로서비스 간 요청 추적

📦 설치

pip install log-collector

개발 의존성 포함:

pip install log-collector[dev]

🚀 빠른 시작

기본 사용법

from log_collector import AsyncLogClient

# 클라이언트 초기화
logger = AsyncLogClient("http://localhost:8000")

# 로그 전송
logger.info("Application started")
logger.warn("High memory usage detected", memory_mb=512)
logger.error("Database connection failed", db_host="localhost")

# 앱 종료 시 자동으로 큐 flush됨

환경 변수 자동 로드

.env 파일 또는 환경 변수로 설정:

LOG_SERVER_URL=http://localhost:8000
SERVICE_NAME=payment-api
NODE_ENV=production
SERVICE_VERSION=v1.2.3
LOG_TYPE=BACKEND

# 환경 변수에서 자동으로 로드됨
logger = AsyncLogClient()

🎯 Feature 1: 자동 호출 위치 추적

모든 로그에 function_name, file_path 자동 포함!

def process_payment(amount):
    logger.info("Processing payment", amount=amount)
    # → function_name="process_payment", file_path="/app/payment.py" 자동 포함!

# 비활성화도 가능
logger.log("INFO", "Manual log", auto_caller=False)

PostgreSQL 분석:

SELECT function_name, COUNT(*) as call_count
FROM logs
WHERE created_at > NOW() - INTERVAL '1 hour'
GROUP BY function_name
ORDER BY call_count DESC;

🌐 Feature 2: HTTP 컨텍스트 자동 수집

웹 프레임워크 환경에서 path, method, ip 자동 포함!

Flask 통합

from flask import Flask, request
from log_collector import AsyncLogClient

app = Flask(__name__)
logger = AsyncLogClient("http://localhost:8000")

@app.before_request
def set_log_context():
    AsyncLogClient.set_request_context(
        path=request.path,
        method=request.method,
        ip=request.remote_addr
    )

@app.after_request
def clear_log_context(response):
    logger.info(f"Request completed: {response.status_code}")
    AsyncLogClient.clear_request_context()
    return response

@app.route('/api/users/<user_id>')
def get_user(user_id):
    logger.info(f"Fetching user {user_id}")
    # → path="/api/users/123", method="GET", ip="127.0.0.1" 자동 포함!
    return {"user_id": user_id}

FastAPI 통합

from fastapi import FastAPI, Request
from log_collector import AsyncLogClient

app = FastAPI()
logger = AsyncLogClient("http://localhost:8000")

@app.middleware("http")
async def log_context_middleware(request: Request, call_next):
    AsyncLogClient.set_request_context(
        path=request.url.path,
        method=request.method,
        ip=request.client.host if request.client else None
    )

    try:
        response = await call_next(request)
        logger.info(f"Request completed: {response.status_code}")
        return response
    finally:
        AsyncLogClient.clear_request_context()

@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    logger.info(f"Fetching user {user_id}")
    # → path, method, ip 자동 포함!
    return {"user_id": user_id}

👤 Feature 3: 사용자 컨텍스트 관리

user_id, trace_id, session_id 등을 모든 로그에 자동 포함!

Context Manager 방식 (권장)

# 특정 블록에만 컨텍스트 적용
with AsyncLogClient.user_context(
    user_id="user_123",
    trace_id="trace_xyz",
    session_id="sess_abc"
):
    logger.info("User logged in")
    # → user_id, trace_id, session_id 자동 포함!

    process_payment()
    logger.info("Payment completed")
    # → 하위 함수에서도 자동으로 컨텍스트 유지!

# with 블록 벗어나면 자동 초기화

중첩 컨텍스트 (자동 병합)

# 외부: tenant_id
with AsyncLogClient.user_context(tenant_id="tenant_1"):
    logger.info("Tenant operation")
    # → tenant_id="tenant_1"

    # 내부: user_id 추가
    with AsyncLogClient.user_context(user_id="user_123"):
        logger.info("User operation")
        # → tenant_id="tenant_1", user_id="user_123" 둘 다 포함!

분산 추적 (Distributed Tracing)

import uuid

def handle_request():
    trace_id = str(uuid.uuid4())

    with AsyncLogClient.user_context(trace_id=trace_id, user_id="user_123"):
        logger.info("Request received")
        call_service_a()  # Service A 호출
        call_service_b()  # Service B 호출
        logger.info("Request completed")
        # → 모든 로그가 같은 trace_id로 추적 가능!

PostgreSQL 분석:

-- trace_id로 전체 요청 흐름 추적
SELECT created_at, service, function_name, message, duration_ms
FROM logs
WHERE trace_id = 'your-trace-id'
ORDER BY created_at;

Set/Clear 방식

# 로그인 시
AsyncLogClient.set_user_context(
    user_id="user_123",
    session_id="sess_abc"
)

logger.info("User action")
# → user_id, session_id 자동 포함

# 로그아웃 시
AsyncLogClient.clear_user_context()

🔧 고급 기능

타이머 측정

# 수동 타이머
timer = logger.start_timer()
result = expensive_operation()
logger.end_timer(timer, "INFO", "Operation completed")
# → duration_ms 자동 계산

# 함수 래퍼 (동기/비동기 자동 감지)
result = logger.measure(lambda: expensive_operation())

에러 추적

try:
    risky_operation()
except Exception as e:
    logger.error_with_trace("Operation failed", exception=e)
    # → stack_trace, error_type, function_name, file_path 자동 포함!

수동 Flush

# 중요한 로그를 즉시 전송
logger.flush()

⚙️ 설정 옵션

logger = AsyncLogClient(
    server_url="http://localhost:8000",
    service="payment-api",
    environment="production",
    service_version="v1.2.3",
    log_type="BACKEND",
    batch_size=1000,          # 배치 크기 (기본: 1000)
    flush_interval=1.0,       # Flush 간격 초 (기본: 1.0)
    enable_compression=True   # gzip 압축 (기본: True)
)

📊 성능

• 앱 블로킹: < 0.1ms per log
• 처리량: > 10,000 logs/sec
• 메모리: < 10MB (1000건 큐)
• 압축률: ~70% (100건 이상 시 자동 압축)

🧪 테스트

# 단위 테스트
pytest tests/

# 통합 테스트 (로그 서버 필요)
pytest tests/test_integration.py

# 커버리지
pytest --cov=log_collector tests/

📝 로그 레벨

logger.trace("Trace message")    # TRACE
logger.debug("Debug message")    # DEBUG
logger.info("Info message")      # INFO
logger.warn("Warning message")   # WARN
logger.error("Error message")    # ERROR
logger.fatal("Fatal message")    # FATAL

🔍 PostgreSQL 쿼리 예제

사용자별 로그 조회

SELECT * FROM logs
WHERE user_id = 'user_123'
ORDER BY created_at DESC
LIMIT 100;

에러 발생률

SELECT
    path,
    method,
    COUNT(*) as total_requests,
    COUNT(CASE WHEN level = 'ERROR' THEN 1 END) as errors,
    ROUND(100.0 * COUNT(CASE WHEN level = 'ERROR' THEN 1 END) / COUNT(*), 2) as error_rate
FROM logs
WHERE created_at > NOW() - INTERVAL '1 hour'
GROUP BY path, method
ORDER BY error_rate DESC;

함수별 성능

SELECT
    function_name,
    COUNT(*) as calls,
    AVG(duration_ms) as avg_ms,
    MAX(duration_ms) as max_ms
FROM logs
WHERE duration_ms IS NOT NULL
GROUP BY function_name
ORDER BY avg_ms DESC;

🚨 주의사항

1. 민감한 정보 포함 금지

   # ❌ 절대 안 됨!
   logger.info("Login", password="secret")
   
   # ✅ 식별자만 사용
   logger.info("Login successful", user_id="user_123")
   

2. 과도한 로깅 피하기

   # ❌ 루프 내부에서 과도한 로깅
   for i in range(10000):
       logger.debug(f"Processing {i}")
   
   # ✅ 주요 이벤트만 로깅
   logger.info(f"Batch processing started", count=10000)
   

📚 추가 문서

• HTTP-CONTEXT-GUIDE.md - HTTP 컨텍스트 완전 가이드
• USER-CONTEXT-GUIDE.md - 사용자 컨텍스트 완전 가이드
• FIELD-AUTO-COLLECTION.md - 자동 필드 수집 상세

🤝 기여

기여는 언제나 환영합니다!

📄 라이선스

MIT License - 자유롭게 사용하세요!

---

Made with ❤️ by Log Analysis System Team