Incident Response（事故應變與 Postmortem）

// 健康檢查端點 — 事故偵測的基礎
app.MapGet("/health/ready", async (AppDbContext db, HttpClient paymentClient) =>
{
    var checks = new Dictionary<string, string>();
 
    // 檢查資料庫
    try {
        await db.Database.CanConnectAsync();
        checks["database"] = "healthy";
    } catch {
        checks["database"] = "unhealthy";
    }
 
    // 檢查下游服務
    try {
        var response = await paymentClient.GetAsync("/health");
        checks["payment-service"] = response.IsSuccessStatusCode
            ? "healthy" : "degraded";
    } catch {
        checks["payment-service"] = "unhealthy";
    }
 
    var isHealthy = checks.Values.All(v => v == "healthy");
    return Results.Json(checks, statusCode: isHealthy ? 200 : 503);
});
 
// 自動回滾 — 部署後的 Canary 檢查
public class CanaryChecker
{
    public async Task<bool> IsCanaryHealthy(string version)
    {
        // 檢查新版本的錯誤率
        var errorRate = await GetErrorRate(version, TimeSpan.FromMinutes(5));
        if (errorRate > 0.05) // 5% 以上錯誤率
        {
            await RollbackDeployment(version);
            await NotifyTeam($"Canary failed: {version}, error rate: {errorRate:P}");
            return false;
        }
        return true;
    }
}

// 事故通知自動化（Slack + PagerDuty）
interface Incident {
  id: string;
  severity: "SEV1" | "SEV2" | "SEV3" | "SEV4";
  title: string;
  service: string;
  detectedAt: Date;
  status: "detected" | "investigating" | "mitigating" | "resolved";
}
 
async function createIncident(alert: AlertPayload): Promise<Incident> {
  const incident: Incident = {
    id: `INC-${Date.now()}`,
    severity: classifySeverity(alert),
    title: alert.summary,
    service: alert.labels.service,
    detectedAt: new Date(),
    status: "detected",
  };
 
  // 根據嚴重度通知不同管道
  if (incident.severity === "SEV1" || incident.severity === "SEV2") {
    await pagerduty.triggerIncident(incident);
    await slack.postToChannel("#incidents", formatIncident(incident));
    await statusPage.createIncident(incident);
  } else {
    await slack.postToChannel("#alerts", formatIncident(incident));
  }
 
  return incident;
}
 
// Postmortem 模板
const postmortemTemplate = `
## Incident: {title}
**Severity**: {severity} | **Duration**: {duration} | **Date**: {date}
 
### Timeline
| Time | Event |
|------|-------|
| HH:MM | Alert triggered |
| HH:MM | IC assigned |
| HH:MM | Root cause identified |
| HH:MM | Fix deployed |
| HH:MM | Incident resolved |
 
### Root Cause
{root_cause_analysis}
 
### Impact
- Users affected: {user_count}
- Duration: {duration}
- Revenue impact: {revenue}
 
### Action Items
| Action | Owner | Deadline | Status |
|--------|-------|----------|--------|
| {action} | {owner} | {date} | TODO |
 
### Lessons Learned
**What went well**: {good}
**What could be improved**: {improve}
`;

# Runbook 自動化 — 常見問題的自動處理
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
 
class Severity(Enum):
    SEV1 = 1
    SEV2 = 2
    SEV3 = 3
    SEV4 = 4
 
@dataclass
class Incident:
    id: str
    severity: Severity
    title: str
    service: str
    detected_at: datetime
    timeline: list[dict]
 
# 自動化 Runbook
class RunbookExecutor:
    async def handle_high_error_rate(self, service: str):
        """當錯誤率過高時的自動應變"""
        # Step 1: 收集資訊
        recent_deploys = await self.get_recent_deploys(service)
        error_logs = await self.get_error_logs(service, minutes=10)
 
        # Step 2: 如果最近有部署，自動回滾
        if recent_deploys and recent_deploys[0].age_minutes < 30:
            await self.rollback(service, recent_deploys[0])
            await self.notify(f"Auto-rollback: {service} to {recent_deploys[0].previous_version}")
            return
 
        # Step 3: 如果不是部署導致，通知 on-call
        await self.page_oncall(
            severity=Severity.SEV2,
            message=f"High error rate on {service}, no recent deploy",
            context={"error_logs": error_logs[:10]}
        )
 
    async def handle_high_latency(self, service: str):
        """當延遲過高時的自動應變"""
        # 檢查資源使用率
        cpu = await self.get_cpu_usage(service)
        if cpu > 80:
            await self.scale_up(service, replicas=2)
            await self.notify(f"Auto-scale: {service} +2 replicas (CPU: {cpu}%)")