feat(m3): ADR-074 M3 — Gitea CI/CD 管線失敗 webhook

新增 workflow_run 事件處理:
- GiteaWorkflowRun Pydantic model
- handle_workflow_run() — status/conclusion=failure → TYPE-1 Incident
- 透過 get_incident_service().create_incident_from_signal() 建立告警
- 純通知路徑,不觸發自動修復

2026-04-12 ogt (ADR-074 M3)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-12 15:35:25 +08:00
parent 00a31abb85
commit 3489e05c84

View File

@@ -39,6 +39,7 @@ from pydantic import BaseModel
from src.core.config import settings
from src.core.logging import get_logger
from src.services.gitea_webhook_service import get_gitea_webhook_service
from src.services.incident_service import get_incident_service
logger = get_logger("awoooi.gitea_webhook")
@@ -100,6 +101,17 @@ class GiteaCommit(BaseModel):
modified: list[str] = []
class GiteaWorkflowRun(BaseModel):
"""Gitea Actions Workflow Run 資訊"""
id: int
name: str
status: str # waiting, running, success, failure, cancelled, skipped
conclusion: str | None = None
head_sha: str | None = None
head_branch: str | None = None
html_url: str | None = None
class GiteaWebhookPayload(BaseModel):
"""Gitea Webhook Payload (通用)"""
action: str | None = None # PR: opened, synchronize, etc.
@@ -113,6 +125,8 @@ class GiteaWebhookPayload(BaseModel):
after: str | None = None # current commit SHA
commits: list[GiteaCommit] | None = None
pusher: dict | None = None
# workflow_run 事件 (ADR-074 M3)
workflow_run: GiteaWorkflowRun | None = None
class GiteaWebhookResponse(BaseModel):
@@ -317,6 +331,13 @@ async def handle_gitea_webhook(
delivery_id=x_gitea_delivery,
)
# workflow_run 事件 (ADR-074 M3: CI/CD 管線失敗告警)
elif x_gitea_event == "workflow_run":
return await handle_workflow_run(
payload=payload,
background_tasks=background_tasks,
)
# Ping 事件 (Gitea 測試連線)
elif x_gitea_event == "ping":
return GiteaWebhookResponse(
@@ -469,6 +490,85 @@ async def handle_push(
)
async def handle_workflow_run(
payload: GiteaWebhookPayload,
background_tasks: BackgroundTasks,
) -> GiteaWebhookResponse:
"""
處理 Gitea Actions workflow_run 事件 — ADR-074 M3
只處理 status=failure或 conclusion=failure的管線失敗。
建立 TYPE-1 Incident純通知不自動修復
"""
wf = payload.workflow_run
if not wf:
return GiteaWebhookResponse(
status="error",
message="Missing workflow_run data",
event_type="workflow_run",
)
# 只關心失敗
failed = wf.status == "failure" or wf.conclusion == "failure"
if not failed:
return GiteaWebhookResponse(
status="ignored",
message=f"workflow_run status='{wf.status}' conclusion='{wf.conclusion}' — not a failure",
event_type="workflow_run",
)
repo = payload.repository.full_name
branch = wf.head_branch or "unknown"
sha_short = (wf.head_sha or "")[:8] or "unknown"
run_url = wf.html_url or ""
logger.warning(
"gitea_ci_pipeline_failed",
repo=repo,
workflow=wf.name,
branch=branch,
sha=sha_short,
run_url=run_url,
)
async def _create_ci_incident() -> None:
try:
svc = get_incident_service()
await svc.create_incident_from_signal({
"alert_name": "GiteaCIPipelineFailed",
"severity": "warning",
"source": "gitea",
"fingerprint": f"gitea-ci-{repo}-{branch}-{sha_short}",
"labels": {
"alertname": "GiteaCIPipelineFailed",
"severity": "warning",
"repo": repo,
"workflow": wf.name,
"branch": branch,
"sha": sha_short,
"run_url": run_url,
"notification_type": "TYPE-1",
"alert_category": "infrastructure",
},
"annotations": {
"summary": f"CI 管線失敗:{repo} [{branch}] {wf.name}",
"description": (
f"Gitea Actions 管線 `{wf.name}` 在 `{branch}` ({sha_short}) 失敗。\n{run_url}"
),
},
})
except Exception:
logger.exception("gitea_ci_incident_create_failed", repo=repo, workflow=wf.name)
background_tasks.add_task(_create_ci_incident)
return GiteaWebhookResponse(
status="accepted",
message=f"CI pipeline failure for '{wf.name}' on '{branch}' queued as TYPE-1 incident",
event_type="workflow_run",
)
# =============================================================================
# Query Endpoints
# =============================================================================