Wonseok Jung
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import time
|
|
import functools
|
|
import statistics
|
|
from typing import List, Callable, Any
|
|
|
|
class PerformanceMonitor:
|
|
"""
|
|
모니터링 및 SLO 측정 엔진 (Phase 3: Monitoring Integration)
|
|
모든 병목 지점에 타이밍 래퍼를 삽입하여 실시간 가시성 확보.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.latencies: List[float] = []
|
|
self.slo_threshold_ms = 200.0 # P95 목표: 200ms 이하
|
|
|
|
def track_latency(self, func: Callable):
|
|
"""메서드 실행 시간을 측정하는 데코레이터"""
|
|
@functools.wraps(func)
|
|
async def wrapper(*args, **kwargs):
|
|
start_time = time.perf_counter()
|
|
result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
|
|
latency = (time.perf_counter() - start_time) * 1000
|
|
|
|
self.latencies.append(latency)
|
|
if latency > self.slo_threshold_ms:
|
|
print(f"[SLO Alert] {func.__name__} violated SLO! Latency: {latency:.2f}ms (Goal: {self.slo_threshold_ms}ms)")
|
|
|
|
return result
|
|
return wrapper
|
|
|
|
def get_stats(self) -> dict:
|
|
"""현재까지의 지연 시간 통계 산출 (P95 포함)"""
|
|
if not self.latencies:
|
|
return {"count": 0}
|
|
|
|
stats = {
|
|
"count": len(self.latencies),
|
|
"avg_ms": statistics.mean(self.latencies),
|
|
"max_ms": max(self.latencies),
|
|
"p95_ms": statistics.quantiles(self.latencies, n=20)[18] if len(self.latencies) >= 20 else "N/A"
|
|
}
|
|
return stats
|
|
|
|
def report(self):
|
|
"""정기 성능 보고서 출력"""
|
|
stats = self.get_stats()
|
|
print("\n" + "="*40)
|
|
print("📊 [SYSTEM PERFORMANCE REPORT]")
|
|
print(f"Total Requests: {stats['count']}")
|
|
print(f"Average Latency: {stats.get('avg_ms', 0):.2f}ms")
|
|
print(f"P95 Latency: {stats['p95_ms']}")
|
|
print("="*40 + "\n")
|
|
|
|
monitor = PerformanceMonitor()
|
|
|
|
import asyncio # for decorator awareness
|