CI/CD 자동화와 감사 컴플라이언스

[GitHub PR 머지]
    ↓
[GitHub Actions: 모델 학습 & 평가]
    ↓ (평가 통과 시)
[MLflow: UC Registry에 새 버전 등록]
    ↓
[GitHub Actions: Alias "champion" 업데이트]
    ↓
[Model Serving 엔드포인트: 자동 전환]

# .github/workflows/deploy-model.yml
name: Deploy ML Model

on:
  push:
    branches: [main]
    paths: ['models/**']

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'

      - name: Install dependencies
        run: pip install databricks-sdk mlflow

      - name: Train and register model
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_CLIENT_ID: ${{ secrets.DATABRICKS_SP_CLIENT_ID }}
          DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_SP_CLIENT_SECRET }}
        run: |
          python scripts/train_and_register.py \
            --model-name "ml_catalog.production_models.fraud_detection" \
            --experiment-name "/Shared/fraud-detection-experiment"

      - name: Promote to champion
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_CLIENT_ID: ${{ secrets.DATABRICKS_SP_CLIENT_ID }}
          DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_SP_CLIENT_SECRET }}
        run: |
          python scripts/promote_model.py \
            --model-name "ml_catalog.production_models.fraud_detection" \
            --alias "champion"

# scripts/promote_model.py
import mlflow
import argparse
from databricks.sdk import WorkspaceClient

def promote_model(model_name: str, alias: str):
    """평가를 통과한 최신 모델 버전에 Alias를 설정합니다."""
    client = mlflow.MlflowClient()

    # 최신 버전 조회
    versions = client.search_model_versions(f"name='{model_name}'")
    latest_version = sorted(versions, key=lambda v: int(v.version))[-1]

    # 메트릭 검증 (예: F1-score > 0.85)
    run = mlflow.get_run(latest_version.run_id)
    f1_score = run.data.metrics.get("f1_score", 0)

    if f1_score < 0.85:
        raise ValueError(f"F1 score {f1_score:.3f} is below threshold 0.85. Deployment aborted.")

    # Alias 업데이트 → Model Serving 엔드포인트 자동 전환
    client.set_registered_model_alias(
        name=model_name,
        alias=alias,
        version=latest_version.version
    )
    print(f"Promoted version {latest_version.version} to '{alias}' (F1: {f1_score:.3f})")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--model-name", required=True)
    parser.add_argument("--alias", default="champion")
    args = parser.parse_args()
    promote_model(args.model_name, args.alias)

-- 최근 7일간 모델 접근 로그 조회
SELECT
    event_time,
    user_identity.email AS user,
    action_name,
    request_params,
    response.status_code
FROM system.access.audit
WHERE
    service_name = 'unityCatalog'
    AND action_name IN (
        'getModel', 'createModelVersion', 'updateModelVersion',
        'setRegisteredModelAlias', 'deleteModelVersion'
    )
    AND event_time >= CURRENT_TIMESTAMP - INTERVAL 7 DAYS
ORDER BY event_time DESC;

-- 특정 모델의 배포 이력 추적 (Alias 변경 이력)
SELECT
    event_time,
    user_identity.email AS deployed_by,
    request_params:name::STRING AS model_name,
    request_params:alias::STRING AS alias,
    request_params:version::STRING AS version
FROM system.access.audit
WHERE
    action_name = 'setRegisteredModelAlias'
    AND request_params:name::STRING LIKE '%fraud_detection%'
ORDER BY event_time DESC;

-- 모델 엔드포인트 호출 통계 (이상 접근 탐지)
SELECT
    DATE_TRUNC('hour', event_time) AS hour,
    user_identity.email AS caller,
    COUNT(*) AS call_count
FROM system.access.audit
WHERE
    service_name = 'modelServing'
    AND action_name = 'serveEndpoint'
    AND event_time >= CURRENT_TIMESTAMP - INTERVAL 24 HOURS
GROUP BY 1, 2
HAVING call_count > 1000  -- 비정상적으로 많은 호출 탐지
ORDER BY call_count DESC;

import mlflow

mlflow.set_registry_uri("databricks-uc")
client = mlflow.MlflowClient()

# 1. Workspace Registry의 모델 버전 목록 조회
# (임시로 레거시 URI 사용)
legacy_client = mlflow.MlflowClient(registry_uri="databricks")
versions = legacy_client.search_model_versions("name='legacy_fraud_detection'")

# 2. 각 버전을 UC Registry로 복사
for version in sorted(versions, key=lambda v: int(v.version)):
    source_uri = f"models:/legacy_fraud_detection/{version.version}"

    new_version = mlflow.register_model(
        model_uri=source_uri,
        name="ml_catalog.production_models.fraud_detection"
    )
    print(f"Migrated version {version.version} → UC version {new_version.version}")

# 3. 프로덕션 버전에 champion Alias 설정
client.set_registered_model_alias(
    name="ml_catalog.production_models.fraud_detection",
    alias="champion",
    version=new_version.version
)

-- 잘못된 예 (접근 불가)
GRANT EXECUTE ON MODEL ml_catalog.models.fraud_detection TO `user@company.com`;

-- 올바른 예 (상위 계층 권한 포함)
GRANT USE CATALOG ON CATALOG ml_catalog TO `user@company.com`;
GRANT USE SCHEMA ON SCHEMA ml_catalog.models TO `user@company.com`;
GRANT EXECUTE ON MODEL ml_catalog.models.fraud_detection TO `user@company.com`;

# Model Serving 엔드포인트 호출 권한 (CAN_QUERY)과
# UC 모델 접근 권한 (EXECUTE)은 별도로 관리됩니다.
# 두 권한 모두 없으면 엔드포인트 호출이 실패합니다.

문제: 담당자 퇴사 시 엔드포인트가 비활성화되거나 오류 발생
해결: 서비스 프린시펄 계정으로 엔드포인트 생성 및 관리

# 잘못된 예 - 새 버전 배포 시 엔드포인트 설정 변경 필요
"entity_version": "5"  # 버전 번호 하드코딩

# 올바른 예 - Alias 변경만으로 자동 전환
"entity_version": "champion"  # Alias 사용