Prometheus에서 수집 가능한 커스텀 메트릭 만들기 (Python)

커스텀 메트릭 생성하게 된 계기

기존에는 KEDA를 사용하여 SQS 큐 안에있는 메시지 개수를 보고 오토스케일링을 했었는데,

문제는 파드가 늘어나면 SQS 메시지 수를 파드 수로 나누어 평균 분산 해버리는 구조여서 원하는 방식대로 오토스케일링 기준을 잡기가 어려웠습니다.

이에 따라 개발자분이 만들어둔 SlotManager의 슬롯 상태를 기준으로 오토스케일링을 하려고 메트릭을 만들게 되었습니다.

메트릭 생성

이제 실제로 제가 어떻게 했는지 과정을 말씀드리겠습니다.

Prometheus Client 의존성 추가

먼저 Python 애플리케이션에서 Prometheus 메트릭을 만들기 위해 prometheus-client 라이브러리를 추가하였습니다.

prometheus-client

• 이 라이브러리는, Counter, Gauge, Summary 같은 메트릭 타입을 제공

pyproject.toml
---
prometheus-client = "^0.21.0"

메트릭 모듈 생성

from prometheus_client import Gauge # 앞서 정의한 prometheus_client 라이브러리에서 Gauge 타입을 사용할 수 있도록 선언해줍니다.

# 활성 슬롯 수 (처리 중 + 대기)
SLOTMANAGER_ACTIVE_TOTAL = Gauge(
    "slotmanager_active_total", # 메트릭을 slotmanager_active_total 라는 이름으로 선언
    "Number of active slot entries (processing + available).", # slotmanager_active_total 메트릭에 대한 설명
)
# 처리 중인 슬롯 수
SLOTMANAGER_ACTIVE_PROCESSING = Gauge(
    "slotmanager_active_processing",
    "Number of slots currently processing.",
)
# 남은 슬롯 수
SLOTMANAGER_AVAILABLE_SLOTS = Gauge(
    "slotmanager_available_slots",
    "Number of available slots.",
)
# 슬롯 활용률 (%)
SLOTMANAGER_UTILIZATION_PERCENT = Gauge(
    "slotmanager_utilization_percent",
    "Slot utilization percentage.",
)
# 설정된 최대 슬롯 수
SLOTMANAGER_MAX_SLOTS = Gauge(
    "slotmanager_max_slots",
    "Configured maximum slots.",
)

# SlotManager의 현재 상태(dict)를 받아서 Prometheus 메트릭 값으로 그대로 반영하는 함수
def update_from_status(status: dict) -> None:
    if not status:
        return

    SLOTMANAGER_MAX_SLOTS.set(status.get("max_slots", 0))
    SLOTMANAGER_ACTIVE_TOTAL.set(status.get("active_total", 0))
    SLOTMANAGER_ACTIVE_PROCESSING.set(status.get("active_processing", 0))
    SLOTMANAGER_AVAILABLE_SLOTS.set(status.get("available_slots", 0))
    SLOTMANAGER_UTILIZATION_PERCENT.set(status.get("utilization_percent", 0))

update_from_status 함수는 SlotManager의 현재 상태(dict)를 받아서 각 값들을 Prometheus Gauage에 그대로 반영(set)하는데,

status에 빈 값이 들어오면, 아무것도 안하고 함수룰 종료 시키려고 아래와 같이 방어 코드를 넣어두었습니다.

if not status:
    return

그리고 개발자 분이 개발한 SlotManager 클래스에서,

앞서 생성한 코드가 있는 디렉터리를 파이썬 패키지로 인식시키기 위해 init.py를 추가해줍니다.

SlotManager에서 메트릭 갱신 연결

이제 개발자 분이 만들어둔 SlotManager의 현재 상태를 찍는 함수에서 앞서 생성한 update_from_status 함수에 status 값을 전달 해줍니다.

slot_manager.py
---
from app.service.metrics.slotmanager_metrics import update_from_status

class SlotManager:
    def get_status(self) -> Dict[str, Any]:
        ...
        status ={
            "max_slots": self.max_slots,
            "active_processing": processing_count,
            "active_total": active_count,
            "utilization_percent": utilization,
            "available_slots": self.max_slots - active_count,
            ...
        }

        update_from_status(status)
        return status

/metrics 경로에서 Prometheus 메트릭 수집 할 수 있도록 정의

main.py
---
@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

FastAPI 애플리케이션에 HTTP GET /metrics 엔드포인트를 등록 해주고,

해당 PATH로 접근하면,

현재 Prometheus Registry에 등록된 모든 메트릭을 Prometheus 표준 텍스트 포맷으로 직렬화해서,


올바른 Content-Type 헤더와 함께 HTTP 응답으로 반환 해줍니다.

---

이렇게 생성을 해서 Prometheus에 메트릭을 수집 시켜주기 위해,

Kubernetes에서 ServiceMonitor 리소스를 생성해주고 결과를 확인하면

정상적으로 메트릭이 나오는 것을 볼 수 있습니다.