GET /projects/flafla/consumer.py
FLA — 실시간 메트릭 파이프라인
데이터센터 인프라 상태를 실시간 수집·시각화하는 사내 모니터링 콘솔. 팀 프로젝트이고, 제가 맡은 부분은 메트릭 파이프라인과 시각화 라이브러리였습니다.
- ▸메트릭이 들어와 화면에 닿기까지의 파이프라인 설계
- ▸AIOKafkaConsumer 기반 인프라별 독립 소비 구조
- ▸SSE 푸시 + 시각화 라이브러리(npm 배포)
data-flow
external agents (per infra)
│ publish cpu-status-{id}
▼
┌──────────┐ topic: cpu-status-1
│ Kafka │ topic: cpu-status-2 ...
└────┬─────┘
│ AIOKafkaConsumer (group_id = fla-{id}) ← 인프라마다 독립
▼
┌──────────┐ SSE ┌─────────────┐
│ FastAPI │ ──────▶ │ React 위젯 │ CPU/RAM/GPU 즉시 갱신
└──────────┘ └─────────────┘fla/consumer.py
# 인프라별 Kafka 토픽을 독립 group_id로 소비해 SSE로 흘린다
async def consume_infra(infra_id: str):
consumer = AIOKafkaConsumer(
f"cpu-status-{infra_id}",
bootstrap_servers=KAFKA_BROKERS,
group_id=f"fla-{infra_id}", # 인프라마다 독립 → 수평 분리
enable_auto_commit=True,
)
await consumer.start()
try:
async for msg in consumer:
metric = json.loads(msg.value)
latest[infra_id] = metric # in-memory 최신값 스냅샷
await hub.publish(infra_id, metric) # 구독 중인 SSE로 팬아웃
finally:
await consumer.stop()데이터가 흐르는 길은 이렇습니다. 외부 에이전트가 인프라별 Kafka 토픽(cpu-status-{id})에 메트릭을 발행 → FastAPI 백엔드가 AIOKafkaConsumer로 인프라마다 독립된 group_id로 소비 → SSE로 클라이언트에 푸시 → 프론트가 CPU/RAM/GPU 위젯을 즉시 갱신.
여기서 제가 의식한 설계 포인트는 ‘인프라가 늘어나도 구조가 안 무너지게’ 하는 것이었습니다. 토픽을 인프라 단위로 나누고 컨슈머도 독립 group_id로 분리해 두면, 인프라가 추가될 때 컨슈머·토픽이 자연스럽게 수평으로 갈라집니다. 한 인프라의 메트릭이 다른 인프라 처리에 끼어들지 않도록 격리한 것이 핵심 판단이었습니다.
Kafka를 고른 이유, polling 대신 SSE를 고른 이유(주기적 요청 없이 서버가 변화를 밀어주는 게 실시간 위젯에 맞다고 봤습니다)도 나름대로 따져보고 정했습니다. 거대한 처리량을 받아본 건 아니지만, ‘왜 이 도구를 골랐는지’ 정도는 스스로 설명해 보려 한 프로젝트입니다.
⚠ self-audit: known-issues 18건
WARN 보안 4 — 인증/권한 경계 미흡 항목 카탈로그화
WARN 기능 차단 5 — 엣지 케이스에서 막히는 흐름 기록
WARN 미구현·품질 9 — 다음에 개선할 항목으로 정리
→ 자기 코드의 약점을 직접 찾아 적어둔 결과물(성숙도 신호)