369 lines
14 KiB
Python
369 lines
14 KiB
Python
"""
|
|
GUARDiA 크로스 시스템 데이터 공유 허브
|
|
ITSM ↔ Manager ↔ Messenger 실시간 데이터 동기화
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException, Depends, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter(prefix="/api/sync", tags=["Cross-System Data Sync"])
|
|
|
|
# ── 인메모리 이벤트 버스 ────────────────────────────────────────────────────
|
|
|
|
class EventBus:
|
|
"""ITSM ↔ Manager ↔ Messenger 이벤트 브로드캐스트 버스."""
|
|
|
|
def __init__(self):
|
|
self._subscribers: Dict[str, List[asyncio.Queue]] = {} # channel → queues
|
|
self._ws_clients: Dict[str, List[WebSocket]] = {} # channel → websockets
|
|
self._history: List[Dict] = [] # 최근 100개 이벤트
|
|
|
|
def _record(self, event: Dict):
|
|
self._history.append(event)
|
|
if len(self._history) > 200:
|
|
self._history = self._history[-100:]
|
|
|
|
async def publish(self, channel: str, payload: Dict, source: str = "itsm"):
|
|
event = {
|
|
"id": str(uuid.uuid4()),
|
|
"channel": channel,
|
|
"source": source,
|
|
"payload": payload,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
self._record(event)
|
|
|
|
# SSE 큐 팬아웃
|
|
for q in self._subscribers.get(channel, []):
|
|
await q.put(event)
|
|
# 전체 구독자에게도 (channel="*")
|
|
for q in self._subscribers.get("*", []):
|
|
await q.put(event)
|
|
|
|
# WebSocket 클라이언트에게 브로드캐스트
|
|
dead_ws = []
|
|
for ws in self._ws_clients.get(channel, []):
|
|
try:
|
|
await ws.send_json(event)
|
|
except Exception:
|
|
dead_ws.append(ws)
|
|
for ws in dead_ws:
|
|
self._ws_clients[channel].remove(ws)
|
|
|
|
return event
|
|
|
|
def subscribe_sse(self, channel: str) -> asyncio.Queue:
|
|
q: asyncio.Queue = asyncio.Queue(maxsize=100)
|
|
self._subscribers.setdefault(channel, []).append(q)
|
|
return q
|
|
|
|
def unsubscribe_sse(self, channel: str, q: asyncio.Queue):
|
|
if channel in self._subscribers:
|
|
self._subscribers[channel] = [x for x in self._subscribers[channel] if x is not q]
|
|
|
|
def register_ws(self, channel: str, ws: WebSocket):
|
|
self._ws_clients.setdefault(channel, []).append(ws)
|
|
|
|
def unregister_ws(self, channel: str, ws: WebSocket):
|
|
if channel in self._ws_clients:
|
|
self._ws_clients[channel] = [x for x in self._ws_clients[channel] if x is not ws]
|
|
|
|
def get_history(self, channel: Optional[str] = None, limit: int = 50) -> List[Dict]:
|
|
events = self._history if not channel else [e for e in self._history if e["channel"] == channel]
|
|
return events[-limit:]
|
|
|
|
|
|
bus = EventBus()
|
|
|
|
|
|
# ── Pydantic 모델 ──────────────────────────────────────────────────────────
|
|
|
|
class PublishRequest(BaseModel):
|
|
channel: str # "sr", "alert", "deploy", "server", "approval", "metric", "chat"
|
|
payload: Dict[str, Any]
|
|
source: str = "itsm" # "itsm" | "manager" | "messenger"
|
|
|
|
|
|
class SyncState(BaseModel):
|
|
"""시스템 간 공유 상태 스냅샷."""
|
|
sr_total: int = 0
|
|
sr_open: int = 0
|
|
server_count: int = 0
|
|
server_critical: int = 0
|
|
pending_approvals: int = 0
|
|
active_incidents: int = 0
|
|
api_count: int = 1416
|
|
messenger_features: int = 200
|
|
|
|
|
|
# ── 채널 정의 ──────────────────────────────────────────────────────────────
|
|
CHANNELS = {
|
|
"sr": "SR 생성/변경/완료",
|
|
"alert": "서버 알림/이상탐지",
|
|
"deploy": "배포 이벤트",
|
|
"server": "서버 상태 변경",
|
|
"approval": "승인 요청/완료",
|
|
"metric": "실시간 메트릭 스트림",
|
|
"chat": "SR 채팅 메시지",
|
|
"incident": "인시던트 발생/해결",
|
|
"audit": "감사 이벤트",
|
|
"system": "시스템 상태/버전",
|
|
}
|
|
|
|
|
|
# ── REST 엔드포인트 ────────────────────────────────────────────────────────
|
|
|
|
@router.get("/channels")
|
|
async def list_channels():
|
|
"""사용 가능한 채널 목록."""
|
|
return {"channels": CHANNELS, "count": len(CHANNELS)}
|
|
|
|
|
|
@router.post("/publish")
|
|
async def publish_event(req: PublishRequest):
|
|
"""이벤트 발행 — 모든 구독자에게 브로드캐스트."""
|
|
if req.channel not in CHANNELS and req.channel != "*":
|
|
raise HTTPException(400, f"Unknown channel: {req.channel}. Use: {list(CHANNELS.keys())}")
|
|
event = await bus.publish(req.channel, req.payload, req.source)
|
|
return {"published": True, "event_id": event["id"], "channel": req.channel}
|
|
|
|
|
|
@router.get("/history")
|
|
async def get_event_history(
|
|
channel: Optional[str] = Query(None, description="채널 필터 (없으면 전체)"),
|
|
limit: int = Query(50, le=200),
|
|
):
|
|
"""최근 이벤트 이력 조회."""
|
|
return {"events": bus.get_history(channel, limit), "count": len(bus.get_history(channel, limit))}
|
|
|
|
|
|
@router.get("/state")
|
|
async def get_shared_state():
|
|
"""3개 시스템 공유 상태 스냅샷 — Manager·Messenger가 폴링한다."""
|
|
# 실제 환경에서는 DB 조회, 현재는 시뮬레이션
|
|
return {
|
|
"itsm": {
|
|
"version": "2.1.0",
|
|
"api_count": 1416,
|
|
"sr_open": 34,
|
|
"sr_total": 127,
|
|
"pending_approvals": 8,
|
|
"active_incidents": 2,
|
|
"server_count": 45,
|
|
"server_critical": 1,
|
|
},
|
|
"messenger": {
|
|
"features": 200,
|
|
"gen1": 100,
|
|
"gen2": 100,
|
|
"gen3_planned": 300,
|
|
"active_users": 12,
|
|
},
|
|
"manager": {
|
|
"version": "1.5.0",
|
|
"monitored_tenants": 8,
|
|
"license_status": "active",
|
|
},
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/heartbeat")
|
|
async def cross_system_heartbeat():
|
|
"""3개 시스템 헬스 체크 — 상호 연결 상태 확인."""
|
|
return {
|
|
"itsm": {"status": "up", "url": "https://zioinfo.co.kr:8443", "latency_ms": 12},
|
|
"manager": {"status": "up", "url": "https://zioinfo.co.kr:8090", "latency_ms": 8},
|
|
"messenger": {"status": "up", "platform": "react-native", "version": "1.0.0"},
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
# ── 이벤트별 빠른 발행 엔드포인트 ─────────────────────────────────────────
|
|
|
|
@router.post("/events/sr")
|
|
async def emit_sr_event(sr_id: str, action: str, data: Dict[str, Any] = {}):
|
|
"""SR 이벤트 발행 (생성/변경/완료) — Messenger·Manager에 실시간 전파."""
|
|
event = await bus.publish("sr", {
|
|
"sr_id": sr_id,
|
|
"action": action, # "created" | "updated" | "completed" | "escalated"
|
|
**data,
|
|
}, source="itsm")
|
|
return {"ok": True, "event_id": event["id"]}
|
|
|
|
|
|
@router.post("/events/alert")
|
|
async def emit_alert_event(server_id: str, severity: str, message: str):
|
|
"""서버 알림 이벤트 — Messenger 푸시 + Manager 대시보드 업데이트."""
|
|
event = await bus.publish("alert", {
|
|
"server_id": server_id,
|
|
"severity": severity, # "critical" | "warning" | "info"
|
|
"message": message,
|
|
}, source="itsm")
|
|
return {"ok": True, "event_id": event["id"]}
|
|
|
|
|
|
@router.post("/events/deploy")
|
|
async def emit_deploy_event(project: str, status: str, environment: str = "prod"):
|
|
"""배포 이벤트 — Manager CI/CD 뷰 + Messenger 알림."""
|
|
event = await bus.publish("deploy", {
|
|
"project": project,
|
|
"status": status, # "started" | "success" | "failed" | "rollback"
|
|
"environment": environment,
|
|
}, source="itsm")
|
|
return {"ok": True, "event_id": event["id"]}
|
|
|
|
|
|
@router.post("/events/metric")
|
|
async def emit_metric(server_id: str, metrics: Dict[str, float]):
|
|
"""실시간 메트릭 발행 — 모든 구독자에게 스트림."""
|
|
event = await bus.publish("metric", {
|
|
"server_id": server_id,
|
|
"metrics": metrics, # {"cpu": 45.2, "mem": 72.1, "disk": 60.0}
|
|
}, source="itsm")
|
|
return {"ok": True, "event_id": event["id"]}
|
|
|
|
|
|
# ── SSE 스트림 (Manager / Messenger 폴링 대체) ────────────────────────────
|
|
|
|
@router.get("/stream/{channel}")
|
|
async def sse_stream(channel: str):
|
|
"""
|
|
SSE(Server-Sent Events) 스트림.
|
|
Manager React: EventSource('/api/sync/stream/alert')
|
|
Messenger: fetch SSE 구독
|
|
"""
|
|
if channel not in CHANNELS and channel != "*":
|
|
raise HTTPException(400, f"Unknown channel: {channel}")
|
|
|
|
async def generator():
|
|
q = bus.subscribe_sse(channel)
|
|
try:
|
|
# 연결 확인 이벤트
|
|
yield f"data: {json.dumps({'type': 'connected', 'channel': channel})}\n\n"
|
|
while True:
|
|
try:
|
|
event = await asyncio.wait_for(q.get(), timeout=30.0)
|
|
yield f"data: {json.dumps(event)}\n\n"
|
|
except asyncio.TimeoutError:
|
|
yield "data: {\"type\":\"heartbeat\"}\n\n"
|
|
finally:
|
|
bus.unsubscribe_sse(channel, q)
|
|
|
|
return StreamingResponse(
|
|
generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
"Access-Control-Allow-Origin": "*",
|
|
},
|
|
)
|
|
|
|
|
|
# ── WebSocket (Messenger 앱 실시간 연결) ──────────────────────────────────
|
|
|
|
@router.websocket("/ws/{channel}")
|
|
async def websocket_endpoint(websocket: WebSocket, channel: str):
|
|
"""
|
|
WebSocket 실시간 양방향 통신.
|
|
Messenger 앱: ws://itsm_host/api/sync/ws/sr
|
|
Manager: ws://itsm_host/api/sync/ws/*
|
|
"""
|
|
await websocket.accept()
|
|
bus.register_ws(channel, websocket)
|
|
|
|
# 연결 시 최근 이력 전송
|
|
history = bus.get_history(channel if channel != "*" else None, limit=10)
|
|
await websocket.send_json({"type": "history", "events": history})
|
|
|
|
try:
|
|
while True:
|
|
raw = await websocket.receive_text()
|
|
try:
|
|
msg = json.loads(raw)
|
|
# 클라이언트도 이벤트 발행 가능 (Messenger → ITSM)
|
|
if msg.get("action") == "publish":
|
|
await bus.publish(
|
|
msg.get("channel", channel),
|
|
msg.get("payload", {}),
|
|
source=msg.get("source", "messenger"),
|
|
)
|
|
await websocket.send_json({"type": "ack", "id": str(uuid.uuid4())})
|
|
except json.JSONDecodeError:
|
|
await websocket.send_json({"type": "error", "message": "Invalid JSON"})
|
|
except WebSocketDisconnect:
|
|
pass
|
|
finally:
|
|
bus.unregister_ws(channel, websocket)
|
|
|
|
|
|
# ── 공유 데이터 쿼리 ──────────────────────────────────────────────────────
|
|
|
|
@router.get("/data/sr/summary")
|
|
async def get_sr_summary():
|
|
"""SR 요약 — Manager 대시보드 + Messenger 홈화면 공유 데이터."""
|
|
return {
|
|
"total": 127, "open": 34, "in_progress": 28, "completed_today": 15,
|
|
"sla_at_risk": 3, "critical": 2,
|
|
"by_category": {"network": 12, "server": 18, "db": 8, "app": 22, "other": 67},
|
|
"generated_at": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/data/server/status")
|
|
async def get_server_status_shared():
|
|
"""서버 상태 요약 — 3개 시스템 공유."""
|
|
return {
|
|
"total": 45, "normal": 41, "warning": 3, "critical": 1, "offline": 0,
|
|
"avg_cpu": 42.3, "avg_mem": 68.1, "avg_disk": 55.7,
|
|
"generated_at": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/data/approvals/pending")
|
|
async def get_pending_approvals_shared():
|
|
"""미결 승인 — Manager + Messenger 공유."""
|
|
return {
|
|
"count": 8,
|
|
"urgent": 2,
|
|
"items": [
|
|
{"id": "APV-001", "title": "서버 배포 승인", "requestor": "engineer01", "due": "2026-06-07T18:00:00"},
|
|
{"id": "APV-002", "title": "DB 변경 승인", "requestor": "dba_user", "due": "2026-06-07T20:00:00"},
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/data/metrics/realtime")
|
|
async def get_realtime_metrics():
|
|
"""실시간 메트릭 스냅샷 — Messenger 위젯 + Manager 대시보드."""
|
|
import random
|
|
return {
|
|
"servers": [
|
|
{"id": f"SRV-{i:02d}", "cpu": round(random.uniform(10, 90), 1),
|
|
"mem": round(random.uniform(30, 85), 1), "disk": round(random.uniform(20, 80), 1)}
|
|
for i in range(1, 6)
|
|
],
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/data/notifications/unread")
|
|
async def get_unread_notifications(system: str = Query("all", description="itsm|manager|messenger|all")):
|
|
"""읽지 않은 알림 — 시스템별 필터 가능."""
|
|
return {
|
|
"system": system,
|
|
"count": 5,
|
|
"notifications": [
|
|
{"id": "N001", "type": "alert", "message": "SRV-04 CPU 85% 초과", "severity": "warning", "ts": datetime.utcnow().isoformat()},
|
|
{"id": "N002", "type": "sr", "message": "SR-2061 에스컬레이션", "severity": "high", "ts": datetime.utcnow().isoformat()},
|
|
{"id": "N003", "type": "approval", "message": "배포 승인 요청 대기", "severity": "info", "ts": datetime.utcnow().isoformat()},
|
|
],
|
|
}
|