DORA 指标与平台工程效能度量：用数据驱动 DevOps 改进

# 伪代码：从流水线事件中统计部署频率
from datetime import datetime, timedelta

def calculate_deployment_frequency(deployments: list[dict], days: int = 30) -> dict:
    """
    deployments: [{"env": "prod", "timestamp": "2026-04-01T10:00:00Z", "service": "api", "status": "success"}]
    """
    cutoff = datetime.utcnow() - timedelta(days=days)
    prod_deploys = [
        d for d in deployments
        if d["env"] == "prod"
        and d["status"] == "success"
        and datetime.fromisoformat(d["timestamp"]) > cutoff
    ]

    return {
        "total": len(prod_deploys),
        "per_day": len(prod_deploys) / days,
        "level": classify_deployment_frequency(len(prod_deploys) / days)
    }

def classify_deployment_frequency(per_day: float) -> str:
    if per_day >= 1:
        return "Elite"      # 每天至少一次
    elif per_day >= 1/7:
        return "High"       # 每周至少一次
    elif per_day >= 1/30:
        return "Medium"     # 每月至少一次
    else:
        return "Low"

def calculate_lead_time(commits: list[dict], deployments: list[dict]) -> list[float]:
    """
    commits: [{"sha": "abc123", "timestamp": "2026-04-01T09:00:00Z"}]
    deployments: [{"sha": "abc123", "prod_timestamp": "2026-04-01T10:30:00Z"}]
    """
    sha_to_commit_time = {c["sha"]: c["timestamp"] for c in commits}
    lead_times = []

    for deploy in deployments:
        if deploy["sha"] in sha_to_commit_time:
            commit_time = datetime.fromisoformat(sha_to_commit_time[deploy["sha"]])
            deploy_time = datetime.fromisoformat(deploy["prod_timestamp"])
            lead_time_hours = (deploy_time - commit_time).total_seconds() / 3600
            lead_times.append(lead_time_hours)

    return lead_times

def calculate_change_failure_rate(
    deployments: list[dict],
    incidents: list[dict],
    rollbacks: list[dict],
    window_hours: int = 24
) -> float:
    """
    判断一次部署是否是「失败变更」：
    - 部署后 24 小时内有 P1/P2 Incident
    - 或者触发了回滚
    """
    failure_deploys = set()

    for deploy in deployments:
        deploy_time = datetime.fromisoformat(deploy["timestamp"])
        window_end = deploy_time + timedelta(hours=window_hours)

        # 检查窗口内是否有高优先级 Incident
        for incident in incidents:
            incident_time = datetime.fromisoformat(incident["created_at"])
            if (deploy_time <= incident_time <= window_end
                    and incident["severity"] in ["P1", "P2"]):
                failure_deploys.add(deploy["id"])
                break

    # 加上所有回滚的部署
    for rollback in rollbacks:
        failure_deploys.add(rollback["original_deploy_id"])

    return len(failure_deploys) / len(deployments) if deployments else 0

def calculate_mttr(incidents: list[dict]) -> dict:
    """
    incidents: [{"id": "INC-123", "created_at": "...", "resolved_at": "...", "severity": "P1"}]
    """
    recovery_times = []

    for incident in incidents:
        if not incident.get("resolved_at"):
            continue  # 未解决的不计入
        created = datetime.fromisoformat(incident["created_at"])
        resolved = datetime.fromisoformat(incident["resolved_at"])
        recovery_hours = (resolved - created).total_seconds() / 3600
        recovery_times.append(recovery_hours)

    if not recovery_times:
        return {"mttr_p50": None, "mttr_p90": None}

    recovery_times.sort()
    n = len(recovery_times)
    return {
        "mttr_p50": recovery_times[int(n * 0.5)],
        "mttr_p90": recovery_times[int(n * 0.9)],
        "sample_size": n
    }

# Prometheus Recording Rule：预计算 DORA 指标
groups:
  - name: dora_metrics
    interval: 1h
    rules:
      # 过去 30 天部署次数（每日）
      - record: dora:deployment_frequency:daily
        expr: |
          increase(cicd_deployments_total{env="prod",status="success"}[1d])

      # 变更失败率（7 天滚动窗口）
      - record: dora:change_failure_rate:7d
        expr: |
          (
            increase(cicd_deployments_total{env="prod",status="failed"}[7d])
            + increase(rollbacks_total{env="prod"}[7d])
          )
          /
          increase(cicd_deployments_total{env="prod"}[7d])