[PYTHON] Decorator를 활용한 모델 추론 레이턴시(Latency) 로깅 시스템 설계 : 성능 최적화를 위한 7가지 해결 방법

모델 추론 레이턴시(Latency) 로깅 시스템 설계

 

딥러닝 모델을 실제 서비스 환경(Production)에 배포할 때, 가장 중요한 지표 중 하나는 바로 레이턴시(Latency)입니다. 모델의 정확도가 아무리 높더라도 추론 속도가 느리다면 사용자 경험은 저하될 수밖에 없습니다. 특히 실시간 추천 시스템이나 자율 주행, 금융 트레이딩 시스템에서는 밀리초(ms) 단위의 지연 시간이 비즈니스의 성패를 가릅니다. 본 포스팅에서는 파이썬의 강력한 문법인 데코레이터(Decorator)를 활용하여, 모델 소스 코드를 수정하지 않고도 우아하게 추론 시간을 측정하고 로깅 시스템을 구축하는 전문적인 설계 패턴과 7가지 실무 예제를 다룹니다.

---

1. 왜 데코레이터인가? 추론 로깅 시스템의 핵심 설계 원칙

모델 추론 코드는 핵심 로직(Tensor 연산, 전처리 등)에 집중해야 합니다. 로그를 남기기 위해 모든 함수 내부에 time.time()을 넣는 방식은 코드의 가독성을 해치고 유지보수를 어렵게 만듭니다. 데코레이터를 사용하면 다음과 같은 장점을 얻을 수 있습니다.

• 관심사의 분리(Separation of Concerns): 비즈니스 로직과 모니터링 로직을 완벽히 분리합니다.
• 재사용성: 한 번 작성한 로깅 데코레이터를 여러 모델 클래스나 함수에 즉시 적용할 수 있습니다.
• 비침습적 설계: 기존 함수의 내부 구현을 건드리지 않고 기능을 확장합니다.

---

2. 추론 환경별 로깅 전략 비교

시스템 아키텍처에 따라 레이턴시를 측정하고 저장하는 방식은 달라져야 합니다. 아래 표는 실무에서 주로 사용되는 3가지 방식의 차이점을 요약한 것입니다.

비교 항목	Standard Logging	Prometheus/Pushgateway	ELK Stack (JSON)
주요 용도	단순 디버깅 및 로컬 확인	실시간 대시보드 및 알람	사후 분석 및 대량 로그 검색
데이터 형태	Plain Text	Time-series Metrics	Structured JSON
성능 오버헤드	낮음	매우 낮음 (UDP/Async 가능)	중간 (I/O 부하 고려 필요)
추천 환경	PoC 스테이지	쿠버네티스(K8s) 기반 서빙	대규모 트래픽 분석 시스템

---

3. 실무 적용을 위한 7가지 핵심 예제 (Python Example)

현업 개발자가 즉시 복사하여 프로젝트에 반영할 수 있는 수준의 고도화된 코드 예제들입니다.

Example 1: 기본 밀리초(ms) 측정 데코레이터

import time
import functools
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ModelInference")

def track_latency(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        end_time = time.perf_counter()
        latency = (end_time - start_time) * 1000
        logger.info(f"Method: {func.__name__} | Latency: {latency:.2f}ms")
        return result
    return wrapper

# 사용 예시
@track_latency
def predict(data):
    time.sleep(0.05)  # 모델 추론 시뮬레이션
    return "Result"

Example 2: CUDA 비동기 처리를 고려한 GPU 레이턴시 측정

PyTorch 등 GPU를 사용하는 경우 time.time()은 정확하지 않습니다. GPU 연산이 완료될 때까지 기다리는 동기화 과정이 필요합니다.

import torch

def track_gpu_latency(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if torch.cuda.is_available():
            start_event = torch.cuda.Event(enable_timing=True)
            end_event = torch.cuda.Event(enable_timing=True)
            start_event.record()
            
            result = func(*args, **kwargs)
            
            end_event.record()
            torch.cuda.synchronize()  # GPU 연산 종료 대기
            latency = start_event.elapsed_time(end_event)
            print(f"GPU Latency: {latency:.2f}ms")
        else:
            result = func(*args, **kwargs)
        return result
    return wrapper

Example 3: 배포 환경에 따른 조건부 로깅 데코레이터

운영 환경(Prod)과 개발 환경(Dev)에서 로깅 수준을 다르게 제어해야 할 때 유용합니다.

import os

def smart_logger(env_key="APP_ENV"):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            env = os.getenv(env_key, "dev")
            start = time.perf_counter()
            res = func(*args, **kwargs)
            duration = time.perf_counter() - start
            
            if env == "prod" and duration > 0.5:  # 운영에선 500ms 초과 시에만 경고 로깅
                logger.warning(f"CRITICAL LATENCY: {duration:.4f}s in {func.__name__}")
            elif env == "dev":
                logger.debug(f"DEBUG: {func.__name__} took {duration:.4f}s")
            return res
        return wrapper
    return decorator

Example 4: JSON 형식의 정형 로깅 (ELK Stack 연동용)

import json
import datetime

def json_latency_logger(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        latency = time.perf_counter() - start
        
        log_data = {
            "timestamp": datetime.datetime.utcnow().isoformat(),
            "function": func.__name__,
            "latency_sec": round(latency, 4),
            "status": "success"
        }
        print(json.dumps(log_data))
        return result
    return wrapper

Example 5: 추론 통계(Statistics)를 위한 클래스 기반 데코레이터

단순 로깅을 넘어 누적 평균, 최댓값 등을 관리하는 상태 저장형 데코레이터입니다.

class InferenceStats:
    def __init__(self):
        self.history = []

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            res = func(*args, **kwargs)
            self.history.append(time.perf_counter() - start)
            return res
        return wrapper

    def get_avg(self):
        return sum(self.history) / len(self.history) if self.history else 0

stats_collector = InferenceStats()

@stats_collector
def model_run(x):
    return x * 2

Example 6: 대규모 트래픽을 위한 샘플링 로깅 (Sampling)

모든 요청을 로깅하면 I/O 부하가 발생합니다. 특정 비율(예: 10%)의 요청만 로깅하는 기법입니다.

import random

def sampled_logging(rate=0.1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if random.random() < rate:
                start = time.perf_counter()
                res = func(*args, **kwargs)
                logger.info(f"Sampled Latency: {time.perf_counter()-start:.4f}s")
                return res
            return func(*args, **kwargs)
        return wrapper
    return decorator

Example 7: 다중 레이어 추론(Preprocessing -> Model -> Postprocessing) 추적

def trace_pipeline(pipeline_name):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print(f"--- Pipeline [{pipeline_name}] Started ---")
            start = time.perf_counter()
            result = func(*args, **kwargs)
            print(f"--- Pipeline [{pipeline_name}] Finished: {time.perf_counter()-start:.4f}s ---")
            return result
        return wrapper
    return decorator

---

4. 결론 및 성능 최적화 제언

데코레이터를 통한 로깅 시스템 구축은 모델 서빙의 안정성을 확보하는 첫걸음입니다. 하지만 로깅 자체가 레이턴시를 유발해서는 안 됩니다. 실무에서는 다음 사항을 반드시 고려하십시오.

1. Non-blocking I/O: 로그를 파일이나 데이터베이스에 쓸 때는 비동기 방식으로 처리하여 메인 추론 루프에 영향을 주지 않아야 합니다.
2. Proper Granularity: 너무 세밀한 단위의 함수까지 데코레이터를 붙이면 오히려 오버헤드가 발생하므로, 주요 병목 지점에만 적용하십시오.
3. Cloud-Native Integration: AWS CloudWatch, GCP Stackdriver 등 클라우드 네이티브 서비스와 연결하여 알람 체계를 구축하십시오.

---

출처 및 참고문헌:

• Python Software Foundation. "Data model - Decorators". python.org
• PyTorch Documentation. "Cuda Timing and Events". pytorch.org
• High Performance Python, 2nd Edition by Micha Gorelick and Ian Ozsvald.