콘텐츠로 이동

LogListener

LogListener는 Circuit Breaker의 동작을 관찰하는 가장 간단한 방법을 제공합니다. Python의 표준 logging 모듈에 연결하여 상태 전환을 기록하며, Circuit Breaker가 어떻게 작동하는지에 대한 명확하고 시간순으로 된 보기를 제공합니다.

CircuitBreaker 및 AsyncCircuitBreaker 에서 활용할 수 있습니다.

기본 사용법

LogListener를 Circuit Breaker의 listeners 목록에 추가합니다.

import logging
from fluxgate import CircuitBreaker
from fluxgate.listeners.log import LogListener

# 콘솔 로깅을 위한 기본 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

cb = CircuitBreaker(
    ...,
    listeners=[LogListener(name="payment_api")],
)

Circuit Breaker가 상태를 변경하면 다음과 같은 로그 레코드를 생성합니다.

[2025-01-15 10:30:45] Circuit Breaker 'payment_api' transitioned from CLOSED to OPEN
[2025-01-15 10:31:45] Circuit Breaker 'payment_api' transitioned from OPEN to HALF_OPEN
[2025-01-15 10:31:50] Circuit Breaker 'payment_api' transitioned from HALF_OPEN to CLOSED

커스텀 Logger

LogListener는 기본적으로 logging.getLogger("fluxgate.listeners.log")를 사용합니다. 라우팅이나 포맷팅을 커스터마이즈하려면 직접 logger를 주입하세요.

import logging
from fluxgate import CircuitBreaker
from fluxgate.listeners.log import LogListener

# Circuit Breaker 이벤트 전용 logger 생성
cb_logger = logging.getLogger("myapp.circuit_breaker")
cb_logger.setLevel(logging.INFO)

cb = CircuitBreaker(
    ...,
    listeners=[LogListener(name="payment_api", logger=cb_logger)],
)

커스텀 로그 레벨

level_map을 사용하여 각 상태 전환에 대한 로그 레벨을 커스터마이징할 수 있습니다. 기본적으로 OPEN과 FORCED_OPEN으로의 전환은 WARNING 레벨로, 나머지는 INFO 레벨로 기록됩니다.

import logging
from fluxgate import CircuitBreaker
from fluxgate.listeners.log import LogListener
from fluxgate.state import State

# OPEN 전환을 ERROR로 기록하여 알림 트리거
level_map = {
    "open": logging.ERROR,
    "half_open": logging.WARNING,
    "closed": logging.DEBUG,
}

cb = CircuitBreaker(
    ...,
    listeners=[LogListener(name="payment_api", level_map=level_map)],
)

두 옵션을 함께 사용할 수도 있습니다:

cb_logger = logging.getLogger("myapp.circuit_breaker")
listener = LogListener(name="payment_api", logger=cb_logger, level_map=level_map)

구조화된 (JSON) 로깅 구현

최신 플랫폼에서 더 나은 가시성을 위해 JSON과 같은 구조화된 형식으로 로그를 출력하는 사용자 정의 Listener를 만들 수 있습니다.

import json
import logging
from fluxgate.listeners import Listener
from fluxgate.signal import Signal

class JsonLogListener(Listener):
    def __init__(self, name, logger):
        self.name = name
        self.logger = logger

    def __call__(self, signal: Signal, /) -> None:
        log_data = {
            "message": "Circuit breaker transition",
            "circuit_name": self.name,
            "previous_state": signal.old_state,
            "current_state": signal.new_state,
            "timestamp_utc": signal.timestamp,
        }
        self.logger.info(json.dumps(log_data))

# 사용
json_logger = logging.getLogger("json_logger")
cb_listener = JsonLogListener("payment_api", json_logger)

파일 로깅 구성

Circuit Breaker 로그를 파일로 보내려면 애플리케이션의 로깅 설정에서 적절한 핸들러를 구성합니다. LogListener는 이 구성을 자동으로 사용합니다.

import logging
from logging.handlers import RotatingFileHandler
from fluxgate import CircuitBreaker
from fluxgate.listeners.log import LogListener

# 로테이팅 파일 핸들러 구성
handler = RotatingFileHandler(
    filename='circuit_breaker.log',
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5,
)
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))

# "fluxgate" 로거에 핸들러 추가
fluxgate_logger = logging.getLogger("fluxgate")
fluxgate_logger.addHandler(handler)
fluxgate_logger.setLevel(logging.INFO)

log_listener = LogListener(name="payment_api")

다음 단계