""" GUARDiA ITSM 백그라운드 스케줄러 (APScheduler 3.x / AsyncIOScheduler). 등록된 작업: 00:05 On-Call 자동 로테이션 → 내일 당직 미배정 시 자동 순환 배정 (A-5) 00:10 SSL 인증서 만료 스캔 → D-1/7/30 알림 01:00 계약·EOL 만료 스캔 → D-30/14/7 알림 06:00 PM 자동 생성 → PmSchedule → WorkTimetable 07:00 WBS 지연 감지 (SI) → 완료율 미달 + 예정일 초과 시 이슈 자동 생성 */30 SLA 위반 점검 → 마감 초과 SR 에스컬레이션 (A-2) ──── AI 에이전트 하트비트 (Paperclip 스타일, Ollama 로컬 LLM) ──── */15 INCIDENT_TRIAGE 에이전트 → 미배정 인시던트 자동 분류 0 * * KB_CURATOR 에이전트 → 해결 SR → KB 자동 생성 (매 정시) 30 8 * SSL_WATCHER 에이전트 → SSL 만료 임박 SR 자동 생성 0 8 * WBS_MONITOR 에이전트 → WBS 지연 위험 LLM 분석 0 9 * PM_SUGGESTER 에이전트 → PM 미등록 서버 탐지 ※ 에이전트 하트비트는 tb_agent_config 활성 레코드가 있을 때만 실행됨 ※ 외부 API 호출 없음 — 모든 LLM 추론은 localhost:11434 (Ollama) main.py lifespan에서 start()/stop() 호출. 외부 API 호출 없음 — 모든 처리는 내부 DB + notify 모듈. """ from __future__ import annotations import logging from datetime import date, datetime, timedelta from typing import Optional from uuid import uuid4 logger = logging.getLogger(__name__) # APScheduler 선택적 임포트 (미설치 시 스케줄러 비활성화) try: from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger _APScheduler_available = True except ImportError: _APScheduler_available = False logger.warning( "apscheduler 미설치 — 백그라운드 스케줄러 비활성화. " "pip install apscheduler 로 설치하세요." ) _scheduler: Optional["AsyncIOScheduler"] = None # ── SSL 만료 스캔 ───────────────────────────────────────────────────────────── async def _scan_ssl_expiry() -> None: """매일 00:10 실행 — SSL 만료 임박 서버에 알림 발송.""" from database import SessionLocal from models import Server, SslAlertLevel from sqlalchemy import select, and_ today = date.today() thresholds = [ (1, SslAlertLevel.EXPIRED, "🔴 SSL 인증서 만료됨"), (7, SslAlertLevel.URGENT, "🟠 SSL 인증서 만료 D-7 이내"), (30, SslAlertLevel.WARN, "🟡 SSL 인증서 만료 D-30 이내"), ] try: async with SessionLocal() as db: result = await db.execute( select(Server).where(Server.ssl_expire_date.isnot(None)) ) servers = result.scalars().all() for srv in servers: expire: date = srv.ssl_expire_date days_left = (expire - today).days level = SslAlertLevel.OK msg_prefix = "" for limit, lv, prefix in thresholds: if days_left <= limit: level = lv msg_prefix = prefix break if level == SslAlertLevel.OK: continue # 알림 발송 message = ( f"{msg_prefix}\n" f"서버: {srv.server_name}\n" f"만료일: {expire.isoformat()} (D-{max(days_left, 0)})\n" f"인증서 경로: {srv.ssl_cert_path or '미등록'}\n" f"즉시 갱신 SR을 접수하세요." ) await _push_ops_notify(f"[SSL 만료 경보] {srv.server_name}", message) logger.info( "SSL 만료 경보: server=%s days_left=%d level=%s", srv.server_name, days_left, level, ) except Exception as exc: logger.error("SSL 만료 스캔 오류: %s", exc, exc_info=True) # ── 계약/EOL 만료 스캔 ──────────────────────────────────────────────────────── async def _scan_contract_expiry() -> None: """매일 01:00 실행 — 계약 만료 및 서버 EOL 임박 알림.""" from database import SessionLocal from models import Institution, Server from sqlalchemy import select today = date.today() warn_days = [7, 14, 30] try: async with SessionLocal() as db: # 기관 계약 만료 스캔 insts = (await db.execute( select(Institution).where(Institution.contract_end.isnot(None)) )).scalars().all() for inst in insts: days = (inst.contract_end - today).days if days in warn_days or days == 0: msg = ( f"📋 계약 만료 {f'D-{days}' if days > 0 else '당일'}\n" f"기관: {inst.inst_name}\n" f"계약 종료일: {inst.contract_end.isoformat()}\n" f"계약 갱신 절차를 진행하세요." ) await _push_ops_notify(f"[계약 만료] {inst.inst_name}", msg) # 서버 EOL 스캔 servers = (await db.execute( select(Server).where(Server.eol_date.isnot(None)) )).scalars().all() for srv in servers: days = (srv.eol_date - today).days if days in warn_days or days == 0: msg = ( f"⚠️ 서버 EOL {f'D-{days}' if days > 0 else '당일'}\n" f"서버: {srv.server_name} ({srv.os_type} {srv.os_version})\n" f"EOL 일자: {srv.eol_date.isoformat()}\n" f"교체/마이그레이션 계획을 수립하세요." ) await _push_ops_notify(f"[EOL 경보] {srv.server_name}", msg) except Exception as exc: logger.error("계약/EOL 만료 스캔 오류: %s", exc, exc_info=True) # ── PM 자동 생성 ────────────────────────────────────────────────────────────── async def _auto_generate_pm() -> None: """ 매일 06:00 실행 — PmSchedule.next_scheduled <= 오늘+advance_days 인 항목에 대해 WorkTimetable(work_type=PM) 자동 생성 후 next_scheduled 업데이트. """ from database import SessionLocal from models import PmSchedule, PmFrequency, WorkTimetable from sqlalchemy import select today = date.today() now = datetime.now() try: async with SessionLocal() as db: result = await db.execute( select(PmSchedule).where(PmSchedule.is_active.is_(True)) ) schedules = result.scalars().all() for sched in schedules: # next_scheduled 미설정 → 지금 기준으로 초기 계산 if not sched.next_scheduled: next_dt = _calc_next(sched, now) async with SessionLocal() as db: s = (await db.execute( select(PmSchedule).where(PmSchedule.id == sched.id) )).scalars().first() if s: s.next_scheduled = next_dt await db.commit() continue # advance_days 이내로 다가온 경우 WorkTimetable 생성 days_until = (sched.next_scheduled.date() - today).days if days_until > sched.advance_days: continue # 아직 여유 있음 # 이미 해당 예정일로 WorkTimetable이 있는지 확인 async with SessionLocal() as db: existing = (await db.execute( select(WorkTimetable).where( WorkTimetable.work_type == "PM", WorkTimetable.scheduled_at == sched.next_scheduled, WorkTimetable.inst_id == sched.inst_id, ) )).scalars().first() if existing: continue # 이미 생성됨 # WorkTimetable 생성 wt = WorkTimetable( work_type = "PM", title = f"[자동] {sched.schedule_name} PM", inst_id = sched.inst_id, server_id = sched.server_id, scheduled_at = sched.next_scheduled, content = ( f"정기 PM 자동 생성\n" f"스케줄: {sched.schedule_name} ({sched.frequency})\n" f"담당자: {sched.assignee or '미지정'}" ), result_status = "PENDING", assignee = sched.assignee, reviewer = sched.reviewer, created_by = "scheduler", ) db.add(wt) # next_scheduled 업데이트 next_dt = _calc_next(sched, sched.next_scheduled) sched_obj = (await db.execute( select(PmSchedule).where(PmSchedule.id == sched.id) )).scalars().first() if sched_obj: sched_obj.last_generated = now sched_obj.next_scheduled = next_dt await db.commit() logger.info( "PM 자동 생성: schedule_id=%d name=%s scheduled=%s", sched.id, sched.schedule_name, sched.next_scheduled, ) # D-7 사전 알림 if sched.notify_before and days_until <= 7: await _push_ops_notify( f"[PM 예정] {sched.schedule_name}", f"📅 정기 PM 예정일이 D-{days_until}입니다.\n" f"스케줄: {sched.schedule_name}\n" f"예정일: {sched.next_scheduled.strftime('%Y-%m-%d')}\n" f"담당자: {sched.assignee or '미지정'}", ) except Exception as exc: logger.error("PM 자동 생성 오류: %s", exc, exc_info=True) def _calc_next(sched: "PmSchedule", from_dt: datetime) -> datetime: """PmSchedule 주기에 따라 다음 예정 일시 계산.""" from models import PmFrequency freq = sched.frequency if freq == PmFrequency.WEEKLY: return from_dt + timedelta(weeks=1) if freq == PmFrequency.BIWEEKLY: return from_dt + timedelta(weeks=2) if freq == PmFrequency.MONTHLY: # 매월 day_of_month 일 dom = sched.day_of_month or 1 y, m = from_dt.year, from_dt.month + 1 if m > 12: m -= 12 y += 1 import calendar dom = min(dom, calendar.monthrange(y, m)[1]) return from_dt.replace(year=y, month=m, day=dom, hour=9, minute=0, second=0) if freq == PmFrequency.QUARTERLY: return from_dt + timedelta(days=91) if freq == PmFrequency.SEMIANNUAL: return from_dt + timedelta(days=182) if freq == PmFrequency.ANNUAL: return from_dt.replace(year=from_dt.year + 1) # CUSTOM: cron 표현식 처리 (croniter 사용 시) try: from croniter import croniter if sched.cron_expr: cron = croniter(sched.cron_expr, from_dt) return cron.get_next(datetime) except ImportError: pass return from_dt + timedelta(days=30) # ── WBS 지연 감지 (SI 프로젝트) ────────────────────────────────────────────── async def _scan_wbs_delay() -> None: """ 매일 07:00 실행 — SI 프로젝트의 WBS 리프 항목 중 planned_end < 오늘 AND completion_pct < 100 인 항목 감지 → 1) WbsItem.status = DELAYED 업데이트 2) 동일 항목으로 기존 이슈가 없으면 ProjectIssue 자동 생성 3) 담당 PM에게 알림 발송 """ from database import SessionLocal from models import ( WbsItem, WbsStatus, SiProject, ProjectIssue, IssueType, IssueStatus, ProjectPhase, ) from sqlalchemy import select, and_ from uuid import uuid4 today = date.today() try: async with SessionLocal() as db: # CLOSED가 아닌 활성 SI 프로젝트의 지연 WBS 리프 항목 검색 delayed_items = (await db.execute( select(WbsItem) .join(SiProject, WbsItem.project_id == SiProject.id) .where( WbsItem.is_leaf == True, # noqa: E712 WbsItem.planned_end < today, WbsItem.completion_pct < 100, WbsItem.status != WbsStatus.CANCELLED, SiProject.is_active == True, # noqa: E712 SiProject.phase.notin_([ProjectPhase.CLOSED]), ) )).scalars().all() for item in delayed_items: # 상태 DELAYED로 변경 if item.status != WbsStatus.DELAYED: item.status = WbsStatus.DELAYED # 동일 WBS 항목에 이미 OPEN 이슈가 있는지 확인 existing = (await db.execute( select(ProjectIssue).where( ProjectIssue.project_id == item.project_id, ProjectIssue.wbs_item_id == item.id, ProjectIssue.status.in_([IssueStatus.OPEN, IssueStatus.IN_PROGRESS]), ProjectIssue.issue_type == IssueType.SCHEDULE, ) )).scalars().first() if not existing: delay_days = (today - item.planned_end).days issue = ProjectIssue( issue_id = f"ISS-{datetime.now().strftime('%Y%m%d')}-{str(uuid4())[:5].upper()}", project_id = item.project_id, wbs_item_id = item.id, issue_type = IssueType.SCHEDULE, title = f"[WBS 지연] {item.wbs_code} {item.title}", description = ( f"WBS 항목이 예정일을 {delay_days}일 초과했습니다.\n" f"예정 완료일: {item.planned_end.isoformat()}\n" f"현재 진척률: {item.completion_pct}%\n" f"담당자: {item.assignee or '미지정'}" ), priority = "HIGH" if delay_days > 7 else "MEDIUM", raised_by = "scheduler", assigned_to = item.assignee, impact = f"WBS {item.wbs_code} 일정 지연으로 후속 작업에 영향 가능", ) db.add(issue) logger.warning( "WBS 지연 이슈 자동 생성: project_id=%d wbs=%s delay=%d일", item.project_id, item.wbs_code, delay_days, ) await db.commit() # 지연 항목이 있는 프로젝트별로 알림 발송 if delayed_items: proj_ids = {i.project_id for i in delayed_items} async with SessionLocal() as db: projs = (await db.execute( select(SiProject).where(SiProject.id.in_(proj_ids)) )).scalars().all() for proj in projs: items_for_proj = [i for i in delayed_items if i.project_id == proj.id] msg = ( f"⚠️ WBS 일정 지연 감지\n" f"프로젝트: {proj.project_name} ({proj.project_code})\n" f"지연 항목 수: {len(items_for_proj)}건\n" + "\n".join( f" - {i.wbs_code} {i.title} " f"({(today - i.planned_end).days}일 초과, {i.completion_pct}%)" for i in items_for_proj[:5] ) + ("\n ..." if len(items_for_proj) > 5 else "") ) await _push_ops_notify(f"[WBS 지연] {proj.project_name}", msg) except Exception as exc: logger.error("WBS 지연 감지 오류: %s", exc, exc_info=True) # ── AI 에이전트 하트비트 디스패처 ───────────────────────────────────────────── async def _agent_heartbeat_by_role(role: str) -> None: """ 역할(role)에 해당하는 활성 AgentConfig 를 조회하여 하트비트 실행. AgentConfig 레코드가 없으면 아무것도 하지 않는다 (설치 전 safe). 외부 API 없음 — 모든 LLM 추론은 Ollama localhost:11434. """ try: from database import SessionLocal from models import AgentConfig from sqlalchemy import select from core.agents import get_agent_engine async with SessionLocal() as db: agents = (await db.execute( select(AgentConfig).where( AgentConfig.role == role, AgentConfig.is_active.is_(True), ) )).scalars().all() if not agents: logger.debug("에이전트 하트비트 스킵: role=%s (활성 에이전트 없음)", role) return engine = get_agent_engine() for agent in agents: await engine.run_heartbeat(agent.id) except Exception as exc: logger.error("에이전트 하트비트 오류: role=%s err=%s", role, exc, exc_info=True) # ── G-3: 라이선스 만료 알림 ────────────────────────────────────────────────── async def _license_expiry_check() -> None: """매일 09:00 KST 실행 — 라이선스 만료 임박 시 ADMIN에게 알림 발송.""" try: from database import SessionLocal from routers.license import get_license_status from models import User from sqlalchemy import select async with SessionLocal() as db: status = await get_license_status(db) if not status.get("activated"): return days = status.get("days_remaining", 0) or 0 is_trial = status.get("is_trial", False) edition = status.get("edition", "") label = "체험판" if is_trial else f"{edition} 라이선스" if status.get("expired"): msg = f"[GUARDiA 긴급] {label}이 만료되었습니다. 즉시 갱신하세요. /license" elif days <= 1: msg = f"[GUARDiA 긴급] {label} 만료 D-{days}! 오늘 갱신하세요. /license" elif days <= 7: msg = f"[GUARDiA 경고] {label} 만료 {days}일 전입니다. 갱신을 준비하세요. /license" elif days <= 30: msg = f"[GUARDiA 알림] {label} 만료 {days}일 전입니다. /license" else: return # 30일 초과: 알림 불필요 # ADMIN 사용자에게 알림 async with SessionLocal() as db2: admins = (await db2.execute( select(User).where(User.role == "ADMIN", User.is_active == True) )).scalars().all() for admin in admins: try: from core.notify import create_notification await create_notification( db2, admin.id, "LICENSE_EXPIRY", f"라이선스 갱신 필요 — {label} (D-{days})", msg ) except Exception: pass try: await db2.commit() except Exception: pass await _push_ops_notify(f"[라이선스 갱신 필요] {label}", msg) logger.info("라이선스 만료 알림 발송: label=%s days=%d", label, days) except Exception as exc: logger.error("라이선스 만료 알림 오류: %s", exc, exc_info=True) # ── 내부 알림 헬퍼 ──────────────────────────────────────────────────────────── async def _push_ops_notify(title: str, body: str) -> None: """운영팀 메신저 채널에 알림 발송 (실패해도 스케줄 중단 없음).""" try: from core.notify import send_messenger import os room = os.getenv("MESSENGER_OPS_ROOM", "ops") await send_messenger(room, {"type": "text", "text": f"[GUARDiA 스케줄러]\n{title}\n\n{body}"}) except Exception as exc: logger.warning("스케줄러 알림 발송 실패: %s", exc) # ── 스케줄러 시작/종료 ──────────────────────────────────────────────────────── def get_scheduler() -> Optional["AsyncIOScheduler"]: return _scheduler def start_scheduler() -> None: """FastAPI lifespan에서 호출 — 스케줄러 등록 및 시작.""" global _scheduler if not _APScheduler_available: logger.warning("APScheduler 미설치 — 스케줄러를 시작하지 않습니다.") return _scheduler = AsyncIOScheduler(timezone="Asia/Seoul") _scheduler.add_job( _scan_ssl_expiry, CronTrigger(hour=0, minute=10, timezone="Asia/Seoul"), id="ssl_expiry_scan", name="SSL 인증서 만료 스캔", replace_existing=True, misfire_grace_time=3600, ) _scheduler.add_job( _scan_contract_expiry, CronTrigger(hour=1, minute=0, timezone="Asia/Seoul"), id="contract_expiry_scan", name="계약·EOL 만료 스캔", replace_existing=True, misfire_grace_time=3600, ) _scheduler.add_job( _auto_generate_pm, CronTrigger(hour=6, minute=0, timezone="Asia/Seoul"), id="pm_auto_generate", name="PM 자동 생성", replace_existing=True, misfire_grace_time=3600, ) _scheduler.add_job( _scan_wbs_delay, CronTrigger(hour=7, minute=0, timezone="Asia/Seoul"), id="wbs_delay_scan", name="SI WBS 지연 감지", replace_existing=True, misfire_grace_time=3600, ) # ── AI 에이전트 하트비트 잡 등록 ───────────────────────────────────────── # 각 역할별 고정 하트비트 크론 — AgentConfig 레코드가 있을 때만 실행 _scheduler.add_job( _agent_heartbeat_by_role, CronTrigger(minute="*/15", timezone="Asia/Seoul"), args=["INCIDENT_TRIAGE"], id="agent_incident_triage", name="AI 인시던트 트리아지 (15분)", replace_existing=True, misfire_grace_time=300, ) _scheduler.add_job( _agent_heartbeat_by_role, CronTrigger(hour="*", minute=0, timezone="Asia/Seoul"), args=["KB_CURATOR"], id="agent_kb_curator", name="AI KB 큐레이터 (매 정시)", replace_existing=True, misfire_grace_time=600, ) _scheduler.add_job( _agent_heartbeat_by_role, CronTrigger(hour=8, minute=30, timezone="Asia/Seoul"), args=["SSL_WATCHER"], id="agent_ssl_watcher", name="AI SSL 감시자 (08:30)", replace_existing=True, misfire_grace_time=3600, ) _scheduler.add_job( _agent_heartbeat_by_role, CronTrigger(hour=8, minute=0, timezone="Asia/Seoul"), args=["WBS_MONITOR"], id="agent_wbs_monitor", name="AI WBS 모니터 (08:00)", replace_existing=True, misfire_grace_time=3600, ) _scheduler.add_job( _agent_heartbeat_by_role, CronTrigger(hour=9, minute=0, timezone="Asia/Seoul"), args=["PM_SUGGESTER"], id="agent_pm_suggester", name="AI PM 제안자 (09:00)", replace_existing=True, misfire_grace_time=3600, ) # ── A-5: On-Call 자동 로테이션 (매일 00:05) ────────────────────────────── try: from core.oncall_rotate import auto_rotate_oncall _scheduler.add_job( auto_rotate_oncall, CronTrigger(hour=0, minute=5, timezone="Asia/Seoul"), id="oncall_auto_rotate", name="On-Call 자동 로테이션 (00:05)", replace_existing=True, misfire_grace_time=3600, ) logger.info("On-Call 자동 로테이션 스케줄 등록 완료") except Exception as exc: logger.warning("On-Call 스케줄 등록 실패 (무시): %s", exc) # ── A-2: SLA 타이머 & 자동 에스컬레이션 (30분마다) ─────────────────────── try: from core.sla import check_sla_violations from apscheduler.triggers.interval import IntervalTrigger _scheduler.add_job( check_sla_violations, IntervalTrigger(minutes=30), id="sla_violation_check", name="SLA 위반 점검 및 에스컬레이션 (30분)", replace_existing=True, misfire_grace_time=300, ) logger.info("SLA 위반 점검 스케줄 등록 완료") except Exception as exc: logger.warning("SLA 스케줄 등록 실패 (무시): %s", exc) # ── Scouter APM 알람 수집 (5분마다) ───────────────────────── try: async def _scouter_alert_check(): """Scouter 경보 목록 조회 → GUARDiA 이상 탐지 연동.""" from core.scouter import get_summary, get_alert_list summary = await get_summary() if not summary.get("enabled"): return # CPU > 80% 또는 에러율 위험 서버 알림 critical = summary.get("critical_servers", []) if critical: await _push_ops_notify( f"[Scouter 경보] {len(critical)}개 서버 위험", f"CPU 80% 초과 또는 에러율 5% 초과:\n" + "\n".join(f" - {s}" for s in critical[:5]) ) _scheduler.add_job( _scouter_alert_check, "interval", minutes=5, id="scouter_alert_check", name="Scouter APM 경보 수집 (5분)", replace_existing=True, misfire_grace_time=60, ) logger.info("Scouter APM 경보 수집 스케줄 등록 완료") except Exception as exc: logger.warning("Scouter 스케줄 등록 실패 (무시): %s", exc) # ── G-3: 라이선스 만료 알림 (매일 09:00 KST) ───────────────────────────── try: _scheduler.add_job( _license_expiry_check, CronTrigger(hour=9, minute=0, timezone="Asia/Seoul"), id="license_expiry_check", name="라이선스 만료 알림 (09:00)", replace_existing=True, misfire_grace_time=3600, ) logger.info("라이선스 만료 알림 스케줄 등록 완료") except Exception as exc: logger.warning("라이선스 만료 알림 스케줄 등록 실패 (무시): %s", exc) # ── Self-Improving Learning Loop (매일 03:00) ──────────────────────────── try: async def _run_lesson_mining(): from database import SessionLocal from core.learning import run_lesson_mining async with SessionLocal() as db: result = await run_lesson_mining(db, days_back=30, min_occurrences=3) logger.info( "학습 마이닝 완료: 패턴=%d 교훈생성=%d 교훈갱신=%d 임계값보정=%s", result.get("patterns_analyzed", 0), result.get("lessons_created", 0), result.get("lessons_updated", 0), result.get("thresholds_calibrated", 0), ) _scheduler.add_job( _run_lesson_mining, CronTrigger(hour=3, minute=0, timezone="Asia/Seoul"), id="learning_lesson_mining", name="학습 패턴 마이닝 & 임계값 보정 (03:00)", replace_existing=True, misfire_grace_time=3600, ) logger.info("학습 루프 스케줄 등록 완료 (매일 03:00)") except Exception as exc: logger.warning("학습 루프 스케줄 등록 실패 (무시): %s", exc) _scheduler.start() logger.info( "GUARDiA 스케줄러 시작 — 등록된 작업: %d개", len(_scheduler.get_jobs()), ) def stop_scheduler() -> None: """FastAPI lifespan 종료 시 호출.""" global _scheduler if _scheduler and _scheduler.running: _scheduler.shutdown(wait=False) logger.info("GUARDiA 스케줄러 종료.") _scheduler = None # ── 배치 잡 동적 APScheduler 등록 / 제거 ──────────────────────────────────── async def _run_batch_job(job_id: int) -> None: """ APScheduler 크론 트리거로 호출되는 배치 잡 실행 함수. tb_batch_job → SSH 명령어 실행 → tb_batch_run 이력 기록. alert_on_fail=True 이면 실패 시 SR 자동 생성. """ from database import SessionLocal from models import BatchJob, BatchRun, BatchRunResult, SRRequest from core.ssh_exec import execute_ssh_command import asyncio async with SessionLocal() as db: job = await db.get(BatchJob, job_id) if not job or job.status != "ACTIVE": return run = BatchRun( job_id=job_id, started_at=datetime.utcnow(), result=BatchRunResult.RUNNING, ) db.add(run) await db.commit() await db.refresh(run) run_id = run.id try: exec_result = await execute_ssh_command( job.server_id, job.command, timeout=job.timeout_sec ) success = exec_result.exit_code == 0 run_result = BatchRunResult.SUCCESS if success else BatchRunResult.FAILED async with SessionLocal() as db: run_obj = await db.get(BatchRun, run_id) if run_obj: run_obj.result = run_result run_obj.exit_code = exec_result.exit_code run_obj.stdout_tail = (exec_result.stdout or "")[-5000:] run_obj.ended_at = datetime.utcnow() await db.commit() # 실패 알림 + SR 자동 생성 if not success and job.alert_on_fail: await _create_sr_for_batch_failure(job, run_id, exec_result.stdout or "") except asyncio.TimeoutError: async with SessionLocal() as db: run_obj = await db.get(BatchRun, run_id) if run_obj: run_obj.result = BatchRunResult.TIMEOUT run_obj.ended_at = datetime.utcnow() run_obj.error_msg = f"타임아웃 ({job.timeout_sec}초 초과)" await db.commit() if job.alert_on_fail: await _create_sr_for_batch_failure(job, run_id, f"타임아웃 ({job.timeout_sec}초)") except Exception as exc: async with SessionLocal() as db: run_obj = await db.get(BatchRun, run_id) if run_obj: run_obj.result = BatchRunResult.FAILED run_obj.ended_at = datetime.utcnow() run_obj.error_msg = str(exc)[:500] await db.commit() logger.error("배치 잡 실행 예외: job_id=%d err=%s", job_id, exc, exc_info=True) finally: # last_run_at / last_result 업데이트 async with SessionLocal() as db: job_obj = await db.get(BatchJob, job_id) run_obj = await db.get(BatchRun, run_id) if job_obj and run_obj: job_obj.last_run_at = run_obj.ended_at or datetime.utcnow() job_obj.last_result = run_obj.result await db.commit() logger.info("배치 잡 완료: job_id=%d run_id=%d", job_id, run_id) async def _create_sr_for_batch_failure(job: Any, run_id: int, stderr_tail: str) -> None: """배치 잡 실패 시 SR 자동 생성 및 운영팀 알림.""" from database import SessionLocal from models import SRRequest try: async with SessionLocal() as db: sr = SRRequest( title=f"[배치 실패] {job.job_name}", description=( f"배치 작업 '{job.job_name}' 실행 실패\n" f"Run ID: {run_id}\n" f"명령어: {job.command}\n\n" f"오류 출력 (마지막 1000자):\n{stderr_tail[-1000:]}" ), sr_type="BATCH_FAIL", priority="HIGH" if job.alert_on_fail else "NORMAL", inst_id=job.inst_id, ) db.add(sr) await db.commit() await db.refresh(sr) sr_id = sr.sr_id await _push_ops_notify( f"[배치 실패] {job.job_name}", f"SR {sr_id} 자동 생성됨\nRun ID: {run_id}\n명령어: {job.command}", ) logger.warning("배치 실패 SR 생성: job=%s sr_id=%s", job.job_name, sr_id) except Exception as exc: logger.error("배치 실패 SR 생성 오류: %s", exc) def enable_batch_job(job_id: int, cron_expr: str) -> bool: """ 배치 잡을 APScheduler에 동적으로 등록한다. POST /api/batch/jobs/{id}/enable 라우터에서 호출. Args: job_id: BatchJob.id cron_expr: "0 2 * * *" 형식 크론 표현식 Returns: True: 등록 성공, False: 스케줄러 미초기화 """ if not _APScheduler_available or not _scheduler: logger.warning("enable_batch_job: 스케줄러 미초기화 (job_id=%d)", job_id) return False job_key = f"batch_{job_id}" try: _scheduler.add_job( _run_batch_job, CronTrigger.from_crontab(cron_expr, timezone="Asia/Seoul"), id=job_key, name=f"배치잡-{job_id}", args=[job_id], replace_existing=True, misfire_grace_time=600, ) logger.info("배치 잡 등록: job_id=%d cron=%s", job_id, cron_expr) return True except Exception as exc: logger.error("배치 잡 등록 실패: job_id=%d err=%s", job_id, exc) return False def disable_batch_job(job_id: int) -> bool: """ 배치 잡을 APScheduler에서 제거한다. POST /api/batch/jobs/{id}/disable 라우터에서 호출. Args: job_id: BatchJob.id Returns: True: 제거 성공, False: 스케줄러 미초기화 또는 잡 없음 """ if not _APScheduler_available or not _scheduler: return False job_key = f"batch_{job_id}" try: _scheduler.remove_job(job_key) logger.info("배치 잡 제거: job_id=%d", job_id) return True except Exception: # 잡이 없는 경우 무시 return False async def init_batch_jobs_from_db() -> None: """ 서버 시작 시 tb_batch_job.status=ACTIVE 인 잡을 자동 등록. start_scheduler() 이후 lifespan에서 호출한다. """ if not _APScheduler_available or not _scheduler: return try: from database import SessionLocal from models import BatchJob, BatchJobStatus from sqlalchemy import select async with SessionLocal() as db: jobs = (await db.execute( select(BatchJob).where(BatchJob.status == BatchJobStatus.ACTIVE) )).scalars().all() registered = 0 for job in jobs: if enable_batch_job(job.id, job.cron_expr): registered += 1 logger.info("배치 잡 자동 등록 완료: %d개", registered) except Exception as exc: logger.error("배치 잡 자동 등록 오류: %s", exc, exc_info=True) # ── 타입 힌트용 Any ──────────────────────────────────────────────────────────── from typing import Any # noqa: E402 — 순환 임포트 방지로 파일 끝에 위치