""" GUARDiA ITSM — On-Call 자동 로테이션 & 에스컬레이션 (Enhancement A-5) 기능: 1. 매일 00:05 스케줄러 실행 → 다음날 당직 미배정 시 자동 배정 2. OncallRotateConfig 의 engineer_list 를 current_index 기준으로 순환 3. 배정 시 메신저/이메일 알림 발송 4. 인시던트/SR 에스컬레이션 체인 실행: 당직자(engineer) → backup_engineer → escalation_to → escalation_chain → ADMIN """ from __future__ import annotations import json import logging from datetime import date, datetime, timedelta from typing import List, Optional logger = logging.getLogger(__name__) # ── 로테이션 설정 조회 헬퍼 ────────────────────────────────────────────────── async def _get_rotate_config(db): """OncallRotateConfig 싱글톤 레코드 조회 (없으면 None).""" from sqlalchemy import select from models import OncallRotateConfig result = await db.execute( select(OncallRotateConfig).where(OncallRotateConfig.id == 1) ) return result.scalars().first() async def get_or_create_rotate_config(db) -> "OncallRotateConfig": """ 설정 레코드가 없으면 기본값으로 생성 후 반환. routers/oncall.py 에서 설정 조회/수정 시 사용. """ from models import OncallRotateConfig cfg = await _get_rotate_config(db) if not cfg: cfg = OncallRotateConfig( id=1, is_active=False, engineer_list="[]", current_index=0, rotate_days=1, default_shift="ALL_DAY", notify_on_assign=True, advance_days=1, ) db.add(cfg) await db.commit() await db.refresh(cfg) return cfg # ── 오늘/내일 당직자 조회 ──────────────────────────────────────────────────── async def get_oncall_for_date(target_date: date, db) -> List["OncallSchedule"]: """특정 날짜의 OncallSchedule 목록 반환.""" from sqlalchemy import select from models import OncallSchedule result = await db.execute( select(OncallSchedule) .where(OncallSchedule.duty_date == target_date) .order_by(OncallSchedule.shift) ) return result.scalars().all() async def get_current_oncall(db) -> Optional["OncallSchedule"]: """오늘 당직자 (ALL_DAY 우선, 없으면 첫 번째 일정) 반환.""" schedules = await get_oncall_for_date(date.today(), db) if not schedules: return None # ALL_DAY 우선 for s in schedules: if s.shift == "ALL_DAY": return s return schedules[0] # ── 자동 로테이션 배정 ─────────────────────────────────────────────────────── async def auto_rotate_oncall() -> None: """ 스케줄러에서 매일 00:05 호출. 로직: 1. OncallRotateConfig.is_active == True 확인 2. (오늘 + advance_days) 날짜에 당직 미배정 확인 3. engineer_list[current_index] 를 당직자로 OncallSchedule 등록 4. current_index = (current_index + 1) % len(engineer_list) 업데이트 5. notify_on_assign == True 이면 메신저 알림 발송 """ from database import SessionLocal from models import OncallSchedule, OncallRotateConfig from sqlalchemy import select try: async with SessionLocal() as db: cfg = await _get_rotate_config(db) if not cfg or not cfg.is_active: logger.debug("On-Call 자동 로테이션 비활성화 상태 — 스킵") return engineers: List[str] = json.loads(cfg.engineer_list or "[]") if not engineers: logger.debug("On-Call 로테이션 엔지니어 목록 비어있음 — 스킵") return target_date = date.today() + timedelta(days=cfg.advance_days) async with SessionLocal() as db: # 해당 날짜 당직 존재 여부 확인 existing = (await db.execute( select(OncallSchedule).where( OncallSchedule.duty_date == target_date, OncallSchedule.shift == cfg.default_shift, ) )).scalars().first() if existing: logger.debug( "On-Call 자동 로테이션 스킵: %s 날짜 당직 이미 등록됨 (engineer=%s)", target_date.isoformat(), existing.engineer, ) return # 로테이션 배정 idx = cfg.current_index % len(engineers) assigned_engineer = engineers[idx] next_idx = (idx + 1) % len(engineers) # 에스컬레이션 체인에서 backup / escalation 설정 chain: List[str] = json.loads(cfg.escalation_chain or "[]") backup = chain[0] if chain else None escalation = chain[1] if len(chain) > 1 else None schedule = OncallSchedule( duty_date = target_date, shift = cfg.default_shift, engineer = assigned_engineer, backup_engineer = backup, escalation_to = escalation, note = f"[자동 배정] 로테이션 순번 {idx}", created_by = "oncall-rotator", ) db.add(schedule) # current_index 업데이트 cfg_obj = await db.get(OncallRotateConfig, 1) if cfg_obj: cfg_obj.current_index = next_idx cfg_obj.updated_at = datetime.now() await db.commit() await db.refresh(schedule) logger.info( "On-Call 자동 배정 완료: date=%s shift=%s engineer=%s (next_idx=%d)", target_date.isoformat(), cfg.default_shift, assigned_engineer, next_idx, ) # 알림 발송 if cfg.notify_on_assign: await _notify_oncall_assigned( target_date, cfg.default_shift, assigned_engineer, backup ) except Exception as exc: logger.error("On-Call 자동 로테이션 오류: %s", exc, exc_info=True) # ── 에스컬레이션 체인 실행 ──────────────────────────────────────────────────── async def escalate_oncall(incident_title: str, db) -> Optional[str]: """ 인시던트/장애 발생 시 당직자 → 백업 → escalation_to → ADMIN 순으로 에스컬레이션. Returns: 에스컬레이션 대상 username 또는 None """ from models import User, UserRole from sqlalchemy import select schedule = await get_current_oncall(db) if schedule: # 1차: 당직자 본인 primary = schedule.engineer # 2차: 백업 담당자 backup = schedule.backup_engineer # 3차: 에스컬레이션 지정 담당자 escalation = schedule.escalation_to else: primary = backup = escalation = None # 4차: 로테이션 설정의 escalation_chain chain_fallback: List[str] = [] cfg = await _get_rotate_config(db) if cfg and cfg.escalation_chain: chain_fallback = json.loads(cfg.escalation_chain) # 5차: ADMIN 사용자 admin_user = (await db.execute( select(User).where( User.role == UserRole.ADMIN, User.is_active.is_(True), ).limit(1) )).scalars().first() admin = admin_user.username if admin_user else None # 순서대로 첫 번째 유효한 대상 찾기 candidates = [primary, backup, escalation] + chain_fallback + ([admin] if admin else []) target = next((c for c in candidates if c), None) if target: msg = ( f"🚨 On-Call 에스컬레이션\n" f"장애/인시던트: {incident_title}\n" f"에스컬레이션 대상: {target}\n" f"일시: {datetime.now().strftime('%Y-%m-%d %H:%M')}" ) try: from core.notify import send_messenger await send_messenger(room="ops", text=msg, title=f"[온콜 에스컬레이션] {incident_title}") except Exception as e: logger.warning("On-Call 에스컬레이션 알림 실패: %s", e) return target # ── 내부 알림 헬퍼 ──────────────────────────────────────────────────────────── async def _notify_oncall_assigned( duty_date: date, shift: str, engineer: str, backup: Optional[str], ) -> None: """당직 자동 배정 알림 발송.""" try: from core.notify import send_messenger msg = ( f"📋 On-Call 자동 배정 알림\n" f"날짜: {duty_date.isoformat()}\n" f"시프트: {shift}\n" f"당직자: {engineer}\n" f"백업: {backup or '미지정'}" ) await send_messenger( room="ops", text=msg, title=f"[온콜 배정] {duty_date.isoformat()} {engineer}", ) except Exception as exc: logger.warning("On-Call 배정 알림 실패 (무시): %s", exc)