guardia-docs/12_확장개발_Priority4_인프라.md
DESKTOP-TKLFCPRython 938b25f286 feat(itsm): G-1~G-12 확장 기능 + 하네스/봇/설치스크립트 구현
G-1: 메신저 Webhook Relay + _send_to_room 실제 httpx 호출 구현
G-2: POST /api/tasks/bulk SR 대량작업 엔드포인트 (최대 100건)
G-3: 라이선스 만료 알림 스케줄러 (매일 09:00 KST)
G-4: 체험판 upgrade_banner 필드 + license.py 배너 로직
G-5: core/auto_rca.py + incidents/problem auto-rca 엔드포인트
G-6: core/deploy_impact.py + vibe impact-analysis 엔드포인트
G-7: core/ticket_classifier.py + SR 생성 시 AI 분류 + ai-suggestion API
G-8: VulnPatchRecord 모델 + vuln_scan 패치추적 4개 엔드포인트
G-9: core/jira_sync.py + gateway Jira/Confluence 연동 엔드포인트
G-10: core/push_notify.py + routers/push.py + PushSubscription 모델
G-11: approvals 다중승인 (위임/서명/기한초과/마감연장)
G-12: alembic.ini + migrations/ + cicd/migrate_to_postgres.sh

하네스: guardia-orchestrator 확장기능 Phase 반영
봇명령어: /sr /status /license /bulk 슬래시 명령어 추가
설치스크립트: setup/ (Ubuntu, CentOS, RHEL, Windows) --test 옵션 포함

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 18:18:52 +09:00

4.5 KiB

GUARDiA ITSM — Priority 4: 인프라 / 플랫폼

문서 버전: 1.0 | 작성일: 2026-05-25


1. PostgreSQL 마이그레이션

database.py 변경

import os
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# 환경변수 기반 DB 자동 선택
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "sqlite+aiosqlite:///./guardia.db"  # 기본: SQLite (개발)
)
# PostgreSQL 예시:
# postgresql+asyncpg://guardia:password@localhost:5432/guardia

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_pre_ping=True,
    pool_size=10 if "postgresql" in DATABASE_URL else 5,
    max_overflow=20 if "postgresql" in DATABASE_URL else 0,
)
SessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Alembic 설정

# 설치
pip install alembic asyncpg

# 초기화
cd C:\GUARDiA\itsm
alembic init migrations

# alembic.ini 수정
# sqlalchemy.url = postgresql+asyncpg://user:pass@localhost:5432/guardia

# 첫 마이그레이션 생성
alembic revision --autogenerate -m "initial_schema"

# 적용
alembic upgrade head

마이그레이션 파일 위치

itsm/
├── alembic.ini
└── migrations/
    ├── env.py
    └── versions/
        └── 001_initial_schema.py

SQLite → PostgreSQL 차이점 주의사항

항목 SQLite PostgreSQL
JSON 컬럼 문자열 저장 JSONB (인덱싱 가능)
날짜 함수 strftime() date_trunc(), extract()
자동 증가 INTEGER PRIMARY KEY SERIAL 또는 BIGSERIAL
동시성 파일 잠금 MVCC (완전 동시 지원)

2. 멀티테넌트 지원

설계 방식: 공유 스키마 + tenant_id 컬럼

# models.py — 주요 테이블에 tenant_id 추가
class Institution(Base):
    tenant_id: str = Column(String(20), nullable=False, index=True)

class SRRequest(Base):
    tenant_id: str = Column(String(20), nullable=False, index=True)

class Server(Base):
    tenant_id: str = Column(String(20), nullable=False, index=True)

JWT에 tenant_id 포함

# core/auth.py
def create_access_token(user: User) -> str:
    payload = {
        "sub": str(user.id),
        "role": user.role,
        "tenant_id": user.inst_code,  # 기관코드를 tenant_id로 사용
        "exp": ...
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

자동 필터링 미들웨어

# 모든 조회 쿼리에 tenant_id 자동 추가
async def get_tenant_db(current_user: User = Depends(get_current_user)):
    async with SessionLocal() as db:
        db.tenant_id = current_user.tenant_id
        yield db

3. RBAC 세분화

신규 역할: DEPLOY_ENGINEER

class UserRole(str, Enum):
    ADMIN = "ADMIN"
    PM = "PM"
    ENGINEER = "ENGINEER"
    DEPLOY_ENGINEER = "DEPLOY_ENGINEER"  # 신규
    CUSTOMER = "CUSTOMER"

권한 매핑

기능 ADMIN PM ENGINEER DEPLOY_ENGINEER CUSTOMER
SR 조회 (자신만)
SR 생성
배포 트리거
배치 즉시 실행
서버 자격증명 조회
에이전트 관리
사용자 관리

구현

# core/auth.py
DEPLOY_ROLES = {UserRole.ADMIN, UserRole.PM, UserRole.DEPLOY_ENGINEER}

def require_deploy_role(current_user: User = Depends(get_current_user)):
    if current_user.role not in DEPLOY_ROLES:
        raise HTTPException(403, "배포 권한이 없습니다")
    return current_user

4. 배포 이력 UI

vibe.html 내 배포 이력 탭 추가

  • Jenkins 빌드 번호별 조회
  • 아티팩트 목록 (다운로드 링크)
  • 롤백 버튼 (이전 빌드로 롤백)

Jenkins 빌드 이력 API (core/cicd.py)

async def get_build_history(project_name: str, limit: int = 10) -> list[BuildInfo]:
    """Jenkins 빌드 이력 조회"""
    url = f"{JENKINS_URL}/job/{project_name}/api/json?tree=builds[number,result,timestamp,duration,artifacts[*]]"
    ...

async def get_artifact_url(project_name: str, build_number: int, artifact: str) -> str:
    """빌드 아티팩트 다운로드 URL 반환"""
    return f"{JENKINS_URL}/job/{project_name}/{build_number}/artifact/{artifact}"