df04254b57
- __init__.py: 按字母順序排列 imports - incident_approval_service.py: 移除未使用 UUID, ApprovalRequest, Incident, IncidentStatus Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
466 lines
16 KiB
Python
466 lines
16 KiB
Python
"""
|
||
Incident-Approval Synchronization Service
|
||
==========================================
|
||
ADR-027: Incident-Approval 同步架構
|
||
|
||
確保 Incident 和 Approval 的原子性同步：
|
||
1. 建立時原子同步 (UnitOfWork + Saga)
|
||
2. 狀態變更時雙向傳播
|
||
|
||
設計原則:
|
||
- PostgreSQL 事務 (UnitOfWork) 確保資料庫原子性
|
||
- Saga Pattern 處理 Redis 寫入失敗
|
||
- 狀態同步 Hook 確保一致性
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-26 (台北時區)
|
||
"""
|
||
|
||
from datetime import UTC, datetime
|
||
from typing import TYPE_CHECKING
|
||
from uuid import uuid4
|
||
|
||
import structlog
|
||
from redis.exceptions import RedisError
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
||
from src.core.constants import (
|
||
APPROVAL_TO_INCIDENT_STATUS,
|
||
APPROVAL_TTL_SECONDS,
|
||
INCIDENT_TTL_SECONDS,
|
||
REDIS_KEY_APPROVAL,
|
||
REDIS_KEY_INCIDENT,
|
||
)
|
||
from src.core.unit_of_work import UnitOfWork
|
||
from src.db.models import ApprovalRecord, IncidentRecord
|
||
from src.models.approval import ApprovalRequestCreate, ApprovalStatus
|
||
|
||
if TYPE_CHECKING:
|
||
from redis.asyncio import Redis
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
|
||
class IncidentApprovalSyncError(Exception):
|
||
"""Incident-Approval 同步失敗"""
|
||
|
||
pass
|
||
|
||
|
||
class IncidentApprovalService:
|
||
"""
|
||
Incident-Approval 同步服務
|
||
|
||
職責:
|
||
1. 原子建立 Incident + Approval (create_with_approval)
|
||
2. 狀態同步 (on_approval_status_change)
|
||
3. TTL 統一管理
|
||
|
||
使用方式:
|
||
service = IncidentApprovalService(session_factory, redis_client)
|
||
|
||
# 原子建立
|
||
incident, approval = await service.create_with_approval(
|
||
incident_data, approval_data
|
||
)
|
||
|
||
# 狀態變更同步
|
||
await service.on_approval_status_change(approval_id, "approved")
|
||
|
||
Warning:
|
||
禁止繞過此 Service 直接建立 Incident 或 Approval
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
redis_client: "Redis",
|
||
):
|
||
"""
|
||
初始化 IncidentApprovalService
|
||
|
||
Args:
|
||
session_factory: SQLAlchemy async session factory
|
||
redis_client: Redis async client
|
||
"""
|
||
self._session_factory = session_factory
|
||
self._redis = redis_client
|
||
|
||
# =========================================================================
|
||
# 原子建立 (Phase 1 Core)
|
||
# =========================================================================
|
||
|
||
async def create_with_approval(
|
||
self,
|
||
incident_data: dict,
|
||
approval_data: ApprovalRequestCreate,
|
||
*,
|
||
link_metadata: bool = True,
|
||
) -> tuple[IncidentRecord, ApprovalRecord]:
|
||
"""
|
||
原子建立 Incident + Approval
|
||
|
||
保證:
|
||
1. PostgreSQL 事務：兩者同時成功或失敗
|
||
2. Saga 補償：Redis 失敗時回滾 PostgreSQL
|
||
|
||
Args:
|
||
incident_data: Incident 資料 (dict for IncidentRecord)
|
||
approval_data: Approval 建立請求
|
||
link_metadata: 是否在 metadata 中建立雙向連結
|
||
|
||
Returns:
|
||
tuple[IncidentRecord, ApprovalRecord]: 建立的記錄
|
||
|
||
Raises:
|
||
IncidentApprovalSyncError: 建立失敗 (已回滾)
|
||
|
||
Warning:
|
||
不要直接使用 incident_service.create() 或 approval_db.create_approval()
|
||
"""
|
||
async with UnitOfWork(self._session_factory) as uow:
|
||
try:
|
||
# 1. 建立 Incident (PostgreSQL)
|
||
incident_id = incident_data.get("incident_id") or str(uuid4())
|
||
incident_data["incident_id"] = incident_id
|
||
|
||
incident_record = IncidentRecord(
|
||
id=incident_id,
|
||
title=incident_data.get("title", ""),
|
||
description=incident_data.get("description", ""),
|
||
severity=incident_data.get("severity", "P2"),
|
||
status=incident_data.get("status", "investigating"),
|
||
source=incident_data.get("source", "alertmanager"),
|
||
fingerprint=incident_data.get("fingerprint"),
|
||
extra_metadata=incident_data.get("metadata", {}),
|
||
)
|
||
uow.session.add(incident_record)
|
||
await uow.flush() # 取得 ID
|
||
|
||
# 2. 建立 Approval (PostgreSQL)
|
||
# 準備 metadata，建立雙向連結
|
||
approval_metadata = approval_data.metadata or {}
|
||
if link_metadata:
|
||
approval_metadata["incident_id"] = incident_id
|
||
|
||
approval_record = ApprovalRecord(
|
||
action=approval_data.action,
|
||
description=approval_data.description,
|
||
status=ApprovalStatus.PENDING,
|
||
risk_level=approval_data.risk_level or "MEDIUM",
|
||
required_signatures=1,
|
||
current_signatures=0,
|
||
signatures=[],
|
||
blast_radius=approval_data.blast_radius.model_dump()
|
||
if approval_data.blast_radius
|
||
else {},
|
||
dry_run_checks=[
|
||
c.model_dump() for c in (approval_data.dry_run_checks or [])
|
||
],
|
||
requested_by=approval_data.requested_by,
|
||
expires_at=approval_data.expires_at,
|
||
extra_metadata=approval_metadata,
|
||
fingerprint=incident_data.get("fingerprint"),
|
||
)
|
||
uow.session.add(approval_record)
|
||
await uow.flush() # 取得 ID
|
||
|
||
# 更新 Incident metadata 連結 Approval
|
||
if link_metadata:
|
||
incident_record.extra_metadata = {
|
||
**(incident_record.extra_metadata or {}),
|
||
"approval_id": approval_record.id,
|
||
}
|
||
|
||
# 3. 寫入 Redis (事務外，需 Saga 補償)
|
||
try:
|
||
await self._write_to_redis(incident_record, approval_record)
|
||
except RedisError as e:
|
||
# Saga 補償: rollback PostgreSQL
|
||
logger.error(
|
||
"redis_write_failed_saga_rollback",
|
||
incident_id=incident_id,
|
||
approval_id=approval_record.id,
|
||
error=str(e),
|
||
)
|
||
await uow.rollback()
|
||
raise IncidentApprovalSyncError(f"Redis write failed: {e}") from e
|
||
|
||
# 4. Commit PostgreSQL
|
||
await uow.commit()
|
||
|
||
logger.info(
|
||
"incident_approval_created",
|
||
incident_id=incident_id,
|
||
approval_id=approval_record.id,
|
||
)
|
||
|
||
return incident_record, approval_record
|
||
|
||
except IncidentApprovalSyncError:
|
||
raise
|
||
except Exception as e:
|
||
logger.exception(
|
||
"incident_approval_create_error",
|
||
error=str(e),
|
||
)
|
||
raise IncidentApprovalSyncError(f"Create failed: {e}") from e
|
||
|
||
async def _write_to_redis(
|
||
self,
|
||
incident: IncidentRecord,
|
||
approval: ApprovalRecord,
|
||
) -> None:
|
||
"""
|
||
寫入 Redis (Working Memory)
|
||
|
||
兩個 key 必須同時成功，使用 pipeline 批量操作。
|
||
|
||
Args:
|
||
incident: Incident 記錄
|
||
approval: Approval 記錄
|
||
"""
|
||
incident_key = f"{REDIS_KEY_INCIDENT}{incident.id}"
|
||
approval_key = f"{REDIS_KEY_APPROVAL}{approval.id}"
|
||
|
||
# 序列化
|
||
incident_json = self._serialize_incident(incident)
|
||
approval_json = self._serialize_approval(approval)
|
||
|
||
# Pipeline 批量寫入
|
||
async with self._redis.pipeline() as pipe:
|
||
pipe.set(incident_key, incident_json, ex=INCIDENT_TTL_SECONDS)
|
||
pipe.set(approval_key, approval_json, ex=APPROVAL_TTL_SECONDS)
|
||
await pipe.execute()
|
||
|
||
logger.debug(
|
||
"redis_written",
|
||
incident_key=incident_key,
|
||
approval_key=approval_key,
|
||
ttl=INCIDENT_TTL_SECONDS,
|
||
)
|
||
|
||
def _serialize_incident(self, record: IncidentRecord) -> str:
|
||
"""序列化 IncidentRecord 為 JSON"""
|
||
import json
|
||
|
||
return json.dumps({
|
||
"incident_id": record.id,
|
||
"title": record.title,
|
||
"description": record.description,
|
||
"severity": record.severity,
|
||
"status": record.status,
|
||
"source": record.source,
|
||
"fingerprint": record.fingerprint,
|
||
"metadata": record.extra_metadata,
|
||
"created_at": record.created_at.isoformat() if record.created_at else None,
|
||
})
|
||
|
||
def _serialize_approval(self, record: ApprovalRecord) -> str:
|
||
"""序列化 ApprovalRecord 為 JSON"""
|
||
import json
|
||
|
||
status_val = record.status.value if hasattr(record.status, "value") else record.status
|
||
risk_val = record.risk_level.value if hasattr(record.risk_level, "value") else record.risk_level
|
||
|
||
return json.dumps({
|
||
"id": record.id,
|
||
"action": record.action,
|
||
"description": record.description,
|
||
"status": status_val,
|
||
"risk_level": risk_val,
|
||
"required_signatures": record.required_signatures,
|
||
"current_signatures": record.current_signatures,
|
||
"signatures": record.signatures,
|
||
"blast_radius": record.blast_radius,
|
||
"requested_by": record.requested_by,
|
||
"created_at": record.created_at.isoformat() if record.created_at else None,
|
||
"metadata": record.extra_metadata,
|
||
})
|
||
|
||
# =========================================================================
|
||
# 狀態同步 (Phase 2)
|
||
# =========================================================================
|
||
|
||
async def on_approval_status_change(
|
||
self,
|
||
approval_id: str,
|
||
new_status: ApprovalStatus | str,
|
||
) -> None:
|
||
"""
|
||
Approval 狀態變更時同步 Incident
|
||
|
||
行為:
|
||
1. 更新 Approval 狀態 (PostgreSQL)
|
||
2. 同步更新關聯 Incident 狀態
|
||
3. 更新 Redis TTL
|
||
|
||
Args:
|
||
approval_id: Approval ID
|
||
new_status: 新狀態 (ApprovalStatus 或字串)
|
||
|
||
Note:
|
||
呼叫時機: sign_approval() 成功後
|
||
"""
|
||
if isinstance(new_status, str):
|
||
new_status = ApprovalStatus(new_status)
|
||
|
||
async with UnitOfWork(self._session_factory) as uow:
|
||
# 1. 取得 Approval
|
||
result = await uow.session.execute(
|
||
select(ApprovalRecord).where(ApprovalRecord.id == approval_id)
|
||
)
|
||
approval = result.scalar_one_or_none()
|
||
|
||
if not approval:
|
||
logger.warning("approval_not_found_for_sync", approval_id=approval_id)
|
||
return
|
||
|
||
# 2. 更新 Approval 狀態
|
||
approval.status = new_status
|
||
if new_status in (ApprovalStatus.APPROVED, ApprovalStatus.REJECTED):
|
||
approval.resolved_at = datetime.now(UTC)
|
||
|
||
# 3. 取得關聯 Incident ID
|
||
incident_id = (approval.extra_metadata or {}).get("incident_id")
|
||
if not incident_id:
|
||
logger.debug(
|
||
"no_linked_incident",
|
||
approval_id=approval_id,
|
||
)
|
||
await uow.commit()
|
||
return
|
||
|
||
# 4. 更新 Incident 狀態
|
||
result = await uow.session.execute(
|
||
select(IncidentRecord).where(IncidentRecord.id == incident_id)
|
||
)
|
||
incident = result.scalar_one_or_none()
|
||
|
||
if incident:
|
||
new_incident_status = APPROVAL_TO_INCIDENT_STATUS.get(
|
||
new_status.value if hasattr(new_status, "value") else new_status,
|
||
"investigating",
|
||
)
|
||
incident.status = new_incident_status
|
||
if new_incident_status == "resolved":
|
||
incident.resolved_at = datetime.now(UTC)
|
||
|
||
logger.info(
|
||
"incident_status_synced",
|
||
incident_id=incident_id,
|
||
approval_id=approval_id,
|
||
new_status=new_incident_status,
|
||
)
|
||
|
||
# 5. 更新 Redis TTL
|
||
try:
|
||
await self._refresh_redis_ttl(incident_id, approval_id)
|
||
except RedisError as e:
|
||
# Redis 失敗不阻斷主流程
|
||
logger.warning(
|
||
"redis_ttl_refresh_failed",
|
||
incident_id=incident_id,
|
||
error=str(e),
|
||
)
|
||
|
||
await uow.commit()
|
||
|
||
async def _refresh_redis_ttl(
|
||
self,
|
||
incident_id: str,
|
||
approval_id: str,
|
||
) -> None:
|
||
"""
|
||
刷新 Redis TTL
|
||
|
||
用於確保關聯的 Incident 和 Approval 同時過期。
|
||
"""
|
||
incident_key = f"{REDIS_KEY_INCIDENT}{incident_id}"
|
||
approval_key = f"{REDIS_KEY_APPROVAL}{approval_id}"
|
||
|
||
async with self._redis.pipeline() as pipe:
|
||
pipe.expire(incident_key, INCIDENT_TTL_SECONDS)
|
||
pipe.expire(approval_key, APPROVAL_TTL_SECONDS)
|
||
await pipe.execute()
|
||
|
||
# =========================================================================
|
||
# 查詢輔助
|
||
# =========================================================================
|
||
|
||
async def get_incident_by_approval_id(
|
||
self,
|
||
approval_id: str,
|
||
) -> IncidentRecord | None:
|
||
"""
|
||
透過 Approval ID 取得關聯的 Incident
|
||
|
||
Args:
|
||
approval_id: Approval ID
|
||
|
||
Returns:
|
||
IncidentRecord | None
|
||
"""
|
||
async with UnitOfWork(self._session_factory) as uow:
|
||
# 取得 Approval
|
||
result = await uow.session.execute(
|
||
select(ApprovalRecord).where(ApprovalRecord.id == approval_id)
|
||
)
|
||
approval = result.scalar_one_or_none()
|
||
|
||
if not approval:
|
||
return None
|
||
|
||
# 取得 Incident ID
|
||
incident_id = (approval.extra_metadata or {}).get("incident_id")
|
||
if not incident_id:
|
||
return None
|
||
|
||
# 取得 Incident
|
||
result = await uow.session.execute(
|
||
select(IncidentRecord).where(IncidentRecord.id == incident_id)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
# =============================================================================
|
||
# Factory Function (Dependency Injection)
|
||
# =============================================================================
|
||
|
||
_incident_approval_service: IncidentApprovalService | None = None
|
||
|
||
|
||
def get_incident_approval_service() -> IncidentApprovalService:
|
||
"""
|
||
取得 IncidentApprovalService 單例
|
||
|
||
使用方式:
|
||
service = get_incident_approval_service()
|
||
await service.on_approval_status_change(approval_id, "approved")
|
||
|
||
Note:
|
||
- 使用全域 session factory 和 redis client
|
||
- 首次呼叫時初始化，後續重用
|
||
"""
|
||
global _incident_approval_service
|
||
|
||
if _incident_approval_service is None:
|
||
from src.core.redis_client import get_redis
|
||
from src.db.base import get_session_factory
|
||
|
||
_incident_approval_service = IncidentApprovalService(
|
||
session_factory=get_session_factory(),
|
||
redis_client=get_redis(),
|
||
)
|
||
|
||
return _incident_approval_service
|
||
|
||
|
||
def reset_incident_approval_service() -> None:
|
||
"""
|
||
重置 IncidentApprovalService 單例 (用於測試)
|
||
"""
|
||
global _incident_approval_service
|
||
_incident_approval_service = None
|