102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
"""
|
|
세션/워커 생성·해제·스케일 이벤트를 기록하고 대시보드로 전달
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import json
|
|
from collections import deque
|
|
from typing import Dict, Any, List
|
|
from threading import Lock
|
|
|
|
LOG_DIR = "logs"
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
|
|
SESSION_EVENT_LOG_PATH = os.path.join(LOG_DIR, "session_events.jsonl")
|
|
SESSION_EVENT_MAX_BYTES = 5 * 1024 * 1024 # 5MB
|
|
SESSION_EVENT_BACKUP_COUNT = 3
|
|
|
|
# 메모리 내 최근 이벤트 버퍼 (대시보드 실시간 전송용)
|
|
_recent_events: deque = deque(maxlen=200)
|
|
_events_lock = Lock()
|
|
|
|
|
|
def _rotate_if_needed() -> None:
|
|
try:
|
|
if os.path.exists(SESSION_EVENT_LOG_PATH) and os.path.getsize(SESSION_EVENT_LOG_PATH) >= SESSION_EVENT_MAX_BYTES:
|
|
ts = time.strftime("%Y%m%d-%H%M%S")
|
|
rotated_path = os.path.join(LOG_DIR, f"session_events_{ts}.jsonl")
|
|
os.replace(SESSION_EVENT_LOG_PATH, rotated_path)
|
|
|
|
rotated = [
|
|
os.path.join(LOG_DIR, f) for f in os.listdir(LOG_DIR)
|
|
if f.startswith("session_events_") and f.endswith(".jsonl")
|
|
]
|
|
rotated.sort(key=lambda p: os.path.getmtime(p), reverse=True)
|
|
for old in rotated[SESSION_EVENT_BACKUP_COUNT:]:
|
|
try:
|
|
os.remove(old)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def log_session_event(
|
|
event_type: str, # "session_create", "session_destroy", "worker_scale_up", "worker_scale_down", "pool_reap", etc.
|
|
model_type: str = "",
|
|
session_id: str = "",
|
|
details: Dict[str, Any] | None = None
|
|
) -> None:
|
|
"""
|
|
세션/워커 이벤트 기록
|
|
- JSONL 파일로 영구 저장 (로테이션)
|
|
- 메모리 버퍼에 최근 이벤트 유지 (대시보드 실시간 전송)
|
|
"""
|
|
record = {
|
|
"timestamp": time.time(),
|
|
"event_type": event_type,
|
|
"model_type": model_type,
|
|
"session_id": session_id,
|
|
"details": details or {}
|
|
}
|
|
|
|
try:
|
|
_rotate_if_needed()
|
|
with open(SESSION_EVENT_LOG_PATH, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
with _events_lock:
|
|
_recent_events.append(record)
|
|
|
|
|
|
def get_recent_events(limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""최근 이벤트 반환 (대시보드 API용)"""
|
|
with _events_lock:
|
|
return list(_recent_events)[-limit:]
|
|
|
|
|
|
def read_events_from_file(limit: int = 200) -> List[Dict[str, Any]]:
|
|
"""파일에서 최근 이벤트 읽기 (초기 로드용)"""
|
|
events = []
|
|
try:
|
|
if not os.path.exists(SESSION_EVENT_LOG_PATH):
|
|
return events
|
|
|
|
with open(SESSION_EVENT_LOG_PATH, "r", encoding="utf-8") as f:
|
|
lines = f.readlines()
|
|
|
|
for line in lines[-limit:]:
|
|
try:
|
|
events.append(json.loads(line.strip()))
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
return events
|
|
|